arborist/aborist/cli.py

"""Aborist CLI: ingest / search / verify / stats."""

from __future__ import annotations

import argparse
import json
import os
import sys
from pathlib import Path

from aborist import __version__
from aborist.ingest import ingest_source, verify_random_sample
from aborist.progress import Progress
from aborist.search import FTS5Backend
from aborist.sources import WikipediaCurDump
from aborist.store import (
    DEFAULT_DB_PATH,
    append_audit,
    connect,
    connect_query,
    stats,
    transaction,
)


def _cmd_ingest(args: argparse.Namespace) -> int:
    if args.source in ("wikipedia_cur", "wikipedia_old"):
        if not args.path:
            print(f"--path is required for {args.source}", file=sys.stderr)
            return 2
        from aborist.sources import WikipediaSqlDump

        table = "cur" if args.source == "wikipedia_cur" else "old"
        shard = None
        if args.shard:
            rank_str, total_str = args.shard.split("/", 1)
            shard = (int(rank_str), int(total_str))
        src = WikipediaSqlDump(path=args.path, table=table, shard=shard)
    elif args.source == "html":  # noqa: SIM114 — keep branch shape
        try:
            from aborist.sources import HtmlPageSource
        except ImportError:
            print(
                "html source requires extras: pip install 'aborist[html]'",
                file=sys.stderr,
            )
            return 2
        urls: list[str] = list(args.url or [])
        if args.urls_from:
            urls.extend(
                line.strip()
                for line in Path(args.urls_from).read_text(encoding="utf-8").splitlines()
                if line.strip() and not line.lstrip().startswith("#")
            )
        if not urls:
            print("html source needs --url or --urls-from", file=sys.stderr)
            return 2
        src = HtmlPageSource(urls, respect_robots=not args.no_robots)
    elif args.source in ("grok_export", "grok_media"):
        if not args.path:
            print(f"--path is required for {args.source}", file=sys.stderr)
            return 2
        from aborist.sources import GrokExportSource, GrokMediaPostsSource

        cls = GrokExportSource if args.source == "grok_export" else GrokMediaPostsSource
        src = cls(path=args.path)
    elif args.source in ("wikipedia_xml", "wikipedia_xml_history", "wikipedia_abstract"):
        if not args.path:
            print(f"--path is required for {args.source}", file=sys.stderr)
            return 2
        from aborist.sources import WikipediaAbstractDump, WikipediaXmlDump

        if args.source == "wikipedia_abstract":
            src = WikipediaAbstractDump(path=args.path)
        else:
            shard = None
            if args.shard:
                rank_str, total_str = args.shard.split("/", 1)
                shard = (int(rank_str), int(total_str))
            src = WikipediaXmlDump(
                path=args.path,
                shard=shard,
                multi_revision=(args.source == "wikipedia_xml_history"),
            )
    elif args.source in ("git_repo", "hg_repo"):
        if not args.path:
            print(f"--path is required for {args.source}", file=sys.stderr)
            return 2
        from aborist.sources import GitRepoSource, MercurialRepoSource

        cls = GitRepoSource if args.source == "git_repo" else MercurialRepoSource
        src = cls(repo_path=args.path)
    elif args.source == "providence":
        # Self-reference: promote STRICT live providence_cache records
        # past the kindergarten window into the document corpus.
        # See docs/self-reference-design.md.
        from aborist.sources.providence import (
            DEFAULT_KINDERGARTEN_SECONDS,
            ProvidenceSource,
        )
        from aborist.store import connect

        # The source reads from the SAME shard it's writing into —
        # promote each shard's own STRICT records to its own
        # documents table. Cross-shard promotion runs as a separate
        # invocation per shard.
        target_db_for_read = args.db
        if args.shards_dir and args.shard:
            rank_str, total_str = args.shard.split("/", 1)
            rank = int(rank_str)
            total = int(total_str)
            digits = max(3, len(str(total - 1)))
            target_db_for_read = Path(args.shards_dir) / f"{rank:0{digits}d}.db"
        if not target_db_for_read:
            print("--db or --shards-dir + --shard required for providence source", file=sys.stderr)
            return 2
        kg_seconds = int(getattr(args, "kindergarten_seconds", None) or DEFAULT_KINDERGARTEN_SECONDS)
        # Open a separate connection for reading; ingest opens its own
        # write connection downstream.
        read_conn = connect(target_db_for_read)
        src = ProvidenceSource(read_conn, kindergarten_seconds=kg_seconds)
    else:
        print(f"unknown source: {args.source}", file=sys.stderr)
        return 2

    # Resolve target DB: if --shards-dir is set with --shard, write to a
    # per-shard file. Each shard owns its own SQLite file, so N parallel
    # ingests have ZERO writer-lock contention.
    target_db = args.db
    if args.shards_dir:
        if not args.shard:
            print(
                "--shards-dir requires --shard rank/total",
                file=sys.stderr,
            )
            return 2
        rank_str, total_str = args.shard.split("/", 1)
        rank = int(rank_str)
        total = int(total_str)
        shards_dir = Path(args.shards_dir)
        shards_dir.mkdir(parents=True, exist_ok=True)
        digits = max(3, len(str(total - 1)))
        target_db = shards_dir / f"{rank:0{digits}d}.db"

    progress: Progress | None = None
    if not args.quiet:
        prefix = ""
        if args.shard:
            prefix = f"[shard {args.shard}] "
        progress = Progress(
            interval=args.progress_interval,
            total_estimate=args.total_estimate,
            prefix=prefix,
        )

    conn = connect(target_db)
    try:
        result = ingest_source(
            conn,
            src,
            chunker_name=args.chunker,
            limit=args.limit,
            batch_size=args.batch_size,
            resume=args.resume,
            progress=progress,
        )
    finally:
        conn.close()
    print(json.dumps(result.__dict__, indent=2, ensure_ascii=False))
    return 0


def _cmd_search(args: argparse.Namespace) -> int:
    conn = (
        connect_query(args.db, shards_dir=args.global_shards_dir)
        if args.global_shards_dir
        else connect(args.db)
    )
    try:
        backend = FTS5Backend(conn)
        hits = backend.search(args.query, limit=args.limit)
    finally:
        conn.close()
    if args.json:
        print(
            json.dumps(
                [
                    {
                        "document_root": h.document_root,
                        "document_uri": h.document_uri,
                        "chunk_idx": h.chunk_idx,
                        "snippet": h.snippet,
                        "score": h.score,
                        "audit_mode": h.audit_mode.value,
                        "title": h.title,
                    }
                    for h in hits
                ],
                indent=2, ensure_ascii=False
            )
        )
    else:
        for h in hits:
            print(f"[{h.audit_mode.value}] {h.score:7.3f}  {h.title or h.document_uri}")
            print(f"    chunk {h.chunk_idx}  root={h.document_root[:16]}…")
            print(f"    {h.snippet}")
            print()
    return 0


def _cmd_verify(args: argparse.Namespace) -> int:
    conn = (
        connect_query(args.db, shards_dir=args.global_shards_dir)
        if args.global_shards_dir
        else connect(args.db)
    )
    try:
        result = verify_random_sample(conn, n=args.n)
    finally:
        conn.close()
    print(json.dumps(result, indent=2, ensure_ascii=False))
    return 0 if result["failed"] == 0 else 1


def _cmd_distill(args: argparse.Namespace) -> int:
    from aborist.distill import get_distiller
    from aborist.distill.runner import distill_existing
    from aborist.store import discover_shards

    try:
        distiller = get_distiller(args.process)
    except ValueError as e:
        print(str(e), file=sys.stderr)
        return 2

    # Sharded mode: iterate over each shard's DB and distill in place.
    # Cores stay in their source shard so the per-shard audit/derivation
    # chains remain self-contained.
    if args.global_shards_dir:
        shard_paths = discover_shards(args.global_shards_dir)
        if not shard_paths:
            print(f"no shards in {args.global_shards_dir}", file=sys.stderr)
            return 2

        per_shard: list[dict] = []
        totals = {
            "scanned": 0,
            "distilled": 0,
            "skipped_existing": 0,
            "skipped_cold": 0,
            "skipped_empty": 0,
        }
        for sp in shard_paths:
            conn = connect(sp)
            try:
                r = distill_existing(
                    conn,
                    distiller,
                    kind=args.kind,
                    source_type=args.source_type,
                    limit=args.limit,
                    chunker_name=args.chunker,
                    batch_size=args.batch_size,
                )
            finally:
                conn.close()
            per_shard.append({"shard": sp.name, **r})
            for k in totals:
                totals[k] += r[k]
        print(json.dumps({**totals, "shards": per_shard}, indent=2, ensure_ascii=False))
        return 0

    conn = connect(args.db)
    try:
        result = distill_existing(
            conn,
            distiller,
            kind=args.kind,
            source_type=args.source_type,
            limit=args.limit,
            chunker_name=args.chunker,
            batch_size=args.batch_size,
        )
    finally:
        conn.close()
    print(json.dumps(result, indent=2, ensure_ascii=False))
    return 0


def _cmd_ask(args: argparse.Namespace) -> int:
    import os

    from aborist.qa import ask
    from aborist.qa.client import OpenAICompatibleClient, StubClient

    base_url = args.endpoint or os.environ.get(
        "ABORIST_LLM_ENDPOINT", "https://hermes.ai.unturf.com/v1"
    )
    model = args.model or os.environ.get(
        "ABORIST_LLM_MODEL",
        "adamo1139/Hermes-3-Llama-3.1-8B-FP8-Dynamic",
    )
    revision = os.environ.get("ABORIST_LLM_REVISION", "")
    quantization = os.environ.get("ABORIST_LLM_QUANTIZATION", "fp8-dynamic")
    api_key = os.environ.get("ABORIST_LLM_API_KEY")

    client: object
    if args.dry_run:
        client = StubClient(
            answer=f"[STUB] would have answered '{args.question}' against root {args.document_root[:16]}…"
        )
    else:
        client = OpenAICompatibleClient(base_url=base_url, api_key=api_key)

    conn = (
        connect_query(args.db, shards_dir=args.global_shards_dir)
        if args.global_shards_dir
        else connect(args.db)
    )
    # Per-call policy override for --answer-mode. Other knobs flow from
    # DEFAULT_POLICY.
    from aborist.qa.runner import DEFAULT_POLICY as _DEFAULT_ASK_POLICY
    call_policy = dict(_DEFAULT_ASK_POLICY)
    if getattr(args, "answer_mode", None):
        call_policy["answer_mode"] = args.answer_mode
    try:
        result = ask(
            conn,
            document_root=args.document_root,
            question=args.question,
            client=client,
            model_id=model,
            revision=revision,
            quantization=quantization,
            policy=call_policy,
        )
    finally:
        conn.close()
    print(json.dumps(result, indent=2, ensure_ascii=False))
    return 0 if result.get("status") in ("cache_hit", "cache_miss_then_written") else 1


def _cmd_query(args: argparse.Namespace) -> int:
    """Multi-source RAG: question -> top-K corpus docs -> Hermes -> cache."""
    import os

    from aborist.qa.client import OpenAICompatibleClient, StubClient
    from aborist.qa.query import query

    base_url = args.endpoint or os.environ.get(
        "ABORIST_LLM_ENDPOINT", "https://hermes.ai.unturf.com/v1"
    )
    model = args.model or os.environ.get(
        "ABORIST_LLM_MODEL",
        "adamo1139/Hermes-3-Llama-3.1-8B-FP8-Dynamic",
    )
    revision = os.environ.get("ABORIST_LLM_REVISION", "")
    quantization = os.environ.get("ABORIST_LLM_QUANTIZATION", "fp8-dynamic")
    api_key = os.environ.get("ABORIST_LLM_API_KEY")

    client: object
    if args.dry_run:
        client = StubClient(
            answer="[STUB] dry-run: would have asked Hermes-3 with the assembled context."
        )
    else:
        client = OpenAICompatibleClient(base_url=base_url, api_key=api_key)

    qa_db = args.qa_db
    if qa_db is None:
        if args.global_shards_dir:
            qa_db = Path(args.global_shards_dir) / "qa.db"
        else:
            qa_db = Path.home() / ".aborist" / "qa.db"
    qa_db = Path(qa_db)

    shards_dir = (
        Path(args.global_shards_dir) if args.global_shards_dir else None
    )
    single_db = None if shards_dir else args.db

    # Apply per-call policy overrides (question_dedup, repair, answer_mode)
    # on top of the default. fidelity is a function-level kwarg, not in
    # the policy.
    from aborist.qa.query import DEFAULT_QUERY_POLICY
    call_policy = dict(DEFAULT_QUERY_POLICY)
    if getattr(args, "question_dedup", None):
        call_policy["question_dedup"] = args.question_dedup
    if getattr(args, "answer_mode", None):
        call_policy["answer_mode"] = args.answer_mode
    if getattr(args, "repair", False):
        # Mechanical-only repair when --repair is set; --repair-reprompts
        # adds the optional re-prompt tier on top. Both default off so
        # `make query` stays single-shot unless a knob is flipped.
        call_policy["repair_enabled"] = True
        call_policy["repair_max_reprompts"] = max(
            0, int(getattr(args, "repair_reprompts", 0))
        )
    # Ticket #000008 Phase 4 — quantifier-guard CLI overrides.
    # Six-level disable hierarchy at Levels 2 (per-call CLI flag)
    # via these flags; Level 3 policy fields are reachable via the
    # underlying policy dict.
    if getattr(args, "no_quantifier_guard", False):
        call_policy["quantifier_guard_enabled"] = False
    if getattr(args, "allow_broad", False):
        # Keeps the classifier on (telemetry stays useful) but
        # zeroes out the apply_caps gate so broad shapes don't
        # get clipped during emergent search.
        call_policy["quantifier_guard_apply_caps"] = False
    if getattr(args, "reject_broad", False):
        # Phase 4 reject-broad: the actual rejection happens inside
        # query() via the policy field; this CLI flag just sets the
        # field. See aborist/qa/query.py for the early-return path.
        call_policy["quantifier_reject_broad"] = True
    if getattr(args, "apply_quantifier_caps", False):
        # Operator opts in to flipping the dry-run gate per-call.
        # Bench-first per §10.11.3 — this flag is the path from
        # dry-run to live-cap.
        call_policy["quantifier_guard_apply_caps"] = True
    # Ticket #000010 — meta-cognition CLI overrides.
    if getattr(args, "no_preflight", False):
        call_policy["metacognition_enabled"] = False
    if getattr(args, "block_on_contradiction", False):
        # Strict mode: hard-block on lexical contradictions instead
        # of label-only.
        call_policy["metacognition_block_on_contradiction"] = True
    if getattr(args, "soft_preflight", False):
        # Ticket #000011 — opt-in to model-assisted soft preflight
        # sidecar. Adds one short LLM round-trip; NEVER gates
        # admissibility (D1 preserved).
        call_policy["soft_preflight_enabled"] = True

    result = query(
        question=args.question,
        qa_db=qa_db,
        chat_client=client,
        model_id=model,
        revision=revision,
        quantization=quantization,
        shards_dir=shards_dir,
        single_db=single_db,
        top_k=args.top_k,
        over_fetch=args.over_fetch,
        max_context_chars=args.max_context_chars,
        policy=call_policy,
        fidelity=getattr(args, "fidelity", None),
        burn_existing=bool(getattr(args, "burn", False)),
        retrieval_keywords=getattr(args, "retrieval_keywords", None),
    )

    # Emit unfirehose-compatible session journal. One JSONL file per
    # `make query` invocation, written to ~/.aborist/unfirehose/{slug}/
    # {session_uuid}.jsonl. Unfirehose's native-harness watcher picks
    # this up automatically (no registration). Failures here must NEVER
    # break the query path — wrap in a broad except & swallow.
    try:
        _emit_query_journal(args.question, result, model)
    except Exception:  # pragma: no cover — best-effort journaling
        pass

    if args.json:
        print(json.dumps(result, indent=2, ensure_ascii=False))
    else:
        print(_render_query_human(result, args.question))
    return (
        0
        if result.get("status") in ("cache_hit", "cache_miss_then_written")
        else 1
    )


def _emit_query_journal(question: str, result: dict, model: str) -> None:
    """Write one unfirehose/1.0 session for this query invocation."""
    from aborist.journal import SessionWriter
    timings = result.get("timings") or {}
    answer = result.get("answer_text") or ""
    aborist_meta = {
        "audit_mode": result.get("audit_mode"),
        "verifier_method": result.get("verifier_method"),
        "n_quotes": result.get("n_quotes"),
        "n_verified": result.get("n_verified"),
        "cache_key": result.get("cache_key"),
        "cache_status": result.get("status"),
        "lookup_path": result.get("lookup_path"),
        "violations": [
            {"kind": v.get("kind"), "rationale": (v.get("rationale") or "")[:160]}
            for v in (result.get("violations") or [])
        ],
        "sources": [
            {"title": s.get("title"), "uri": s.get("document_uri"), "used": s.get("used"), "role": s.get("source_role")}
            for s in (result.get("sources") or [])
        ],
        "timings_ms": timings,
        "answer_mode": (result.get("policy") or {}).get("answer_mode"),
    }
    with SessionWriter(first_prompt=question) as s:
        s.user_message(question)
        s.assistant_message(
            answer,
            model=model,
            provider="hermes",
            stop_reason="end_turn",
            duration_ms=int(timings.get("total_ms") or 0) or None,
            aborist_meta=aborist_meta,
        )


# Soft-demote violation kinds that demote STRICT to HYBRID without
# rejecting the pointer outright. WARRANT_MISSING / TITLE_MISMATCH /
# DEFLECTION_DETECTED handled separately as hard demotes (their
# presence determines POINTER-LINKED vs ANCHOR-WARRANTED).
_SOFT_DEMOTE_VIOLATION_KINDS = frozenset({
    "LAZY_ANCHOR_DEMOTED",
    "POINTER_OVERFLOW_TRIMMED",
    "TOO_MANY_CLAIMS",
    "BARE_NAME_CLAIM",
    # FORMAT_COLLAPSED — model abandoned the claim_lattice_pointer
    # protocol (multi-line prose, zero [E\d+] tags). Soft-demotes
    # STRICT → HYBRID; pairs with the bottom UNGROUNDED rung when the
    # parser found nothing groundable, but at least surfaces the
    # collapse cause to the operator at audit-line glance.
    "FORMAT_COLLAPSED",
    # Ticket #000008 Phase 4 — broad-quantifier soft-demotes (§10.3).
    # Per §10.3 these stay as soft demotes (cap at ANCHOR-WARRANTED)
    # rather than minting a new audit_mode token. The audit-line tail
    # (rendered by _render_warrant_tail) names which one fired so an
    # operator can tell at a glance.
    "BROAD_QUANTIFIER_RUNAWAY",      # raw_line_count >> pointer_count
    "BROAD_QUANTIFIER_CAP_APPLIED",  # preflight cap fired below default
    "BROAD_QUANTIFIER_SCOPE_UNBOUND", # unbounded universal reached the LLM
    # BROAD_QUANTIFIER_REJECTED is a HARD demote (UNGROUNDED via
    # early-return) — listed here for completeness but doesn't
    # belong in the soft-demote set.
})


def _ladder_rung_for_lattice(audit_mode: str, violations: list[dict] | None) -> str:
    """Map (audit_mode, violations) → four-rung ladder for claim-lattice
    methods. Renderer-only transformation; schema column unchanged.

    The ladder names a strictly stronger property at each rung:

        POINTER-LINKED       pointer/source/chunk verified;
                             warrant either didn't apply or failed
        ANCHOR-WARRANTED     pointer-linked AND cited evidence contains
                             required anchors (warrant ran & passed);
                             other soft-demote violations may be present
        EVIDENCE-WARRANTED   anchor-warranted AND no soft demotes
        UNGROUNDED           n_verified == 0 (existing audit_mode)

    HYBRID adds `-PARTIAL` suffix to whichever rung applies.

    The rung logic uses the violations list as the signal — no new
    verifier output field needed. Three discriminators:
      - WARRANT_MISSING in violations → warrant ran and failed at
        least one claim → POINTER-LINKED (pointer ok but warrant
        didn't anchor the claim)
      - any soft-demote kind in violations → ANCHOR-WARRANTED (the
        rung is reached but other demotes pulled it back from STRICT)
      - no warrant miss AND no soft demotes → EVIDENCE-WARRANTED
    """
    if audit_mode == "UNGROUNDED":
        return "UNGROUNDED"
    kinds = {v.get("kind") for v in (violations or [])}
    # Hard demotes: WARRANT_MISSING, TITLE_MISMATCH, DEFLECTION_DETECTED
    # all indicate the citation is structurally misaligned with the
    # claim or the answer is structurally off-topic — pointer resolved
    # but the cited evidence (span, source, or whole answer) doesn't
    # actually support the user's question. All three drop to
    # POINTER-LINKED. (DEFLECTION_DETECTED was previously a soft demote
    # to ANCHOR-WARRANTED; the comeliness/fetish/investitures emergent
    # case showed that "the model totally shifted topic but anchored
    # the new topic" earned ANCHOR-WARRANTED unfairly. Off-topic
    # belongs at the lower rung.)
    if (
        "WARRANT_MISSING" in kinds
        or "TITLE_MISMATCH" in kinds
        or "DEFLECTION_DETECTED" in kinds
    ):
        rung = "POINTER-LINKED"
    elif kinds & _SOFT_DEMOTE_VIOLATION_KINDS:
        rung = "ANCHOR-WARRANTED"
    else:
        rung = "EVIDENCE-WARRANTED"
    if audit_mode == "HYBRID":
        rung = rung + "-PARTIAL"
    return rung


def _render_audit_label(
    audit_mode: str,
    verifier_method: str,
    violations: list[dict] | None = None,
) -> str:
    """Map (audit_mode, verifier_method, violations) → human-readable
    display label.

    Schema-level audit_mode names what the lexical verifier could
    confirm; the display label names what THAT means in honesty
    terms. STRICT in claim_lattice mode = "every pointer resolves
    to a valid evidence object whose source_role is allowed AND
    every cited span passes the citation-overlap coverage check."
    That is NOT full semantic entailment. The display label spells
    out the actual property so users don't read STRICT as "the
    answer is correct."

    Four-rung ladder for claim-lattice methods (per ticket #000005):

        POINTER-LINKED       pointer verified; warrant either didn't
                             apply or failed for some claim
        ANCHOR-WARRANTED     pointer-linked + warrant passed where
                             it ran; other soft demotes may apply
        EVIDENCE-WARRANTED   anchor-warranted + no soft demotes
        UNGROUNDED           no verified pairs

    HYBRID audits get a `-PARTIAL` suffix on whichever rung applies.

    Plus the verifier-method tail (`· via claim_lattice`,
    `· via claim_lattice_pointer`) so the user can see WHICH
    verifier path produced the verdict.

    For quote / span / entity / paraphrase mode the audit_mode
    labels carry less risk of overclaiming (they verify against
    pinned spans, not synthesis) — keep them as-is.

    `violations` defaults to None for backward compatibility with
    callers that don't have access to the violation list. With
    None, the ladder falls back to EVIDENCE-WARRANTED (the most
    optimistic rung) — operators get the same surface as before
    until callers thread violations through.
    """
    is_claim_lattice = verifier_method.startswith("claim_lattice")
    if is_claim_lattice:
        rung = _ladder_rung_for_lattice(audit_mode, violations)
        return f"{rung} · via {verifier_method}"
    # Quote / span / entity / paraphrase: keep audit_mode as the
    # primary token; append method for clarity.
    return f"{audit_mode} · via {verifier_method}"


def _render_warrant_tail(result: dict) -> str:
    """Append a tail to the audit-line label that names the specific
    warrant failure mode when one fired. Surfaces the cap reason at
    the user-facing layer without overloading audit_mode.

    Two failure modes today:
      - WARRANT_MISSING: relation/date/etc. anchor extracted from
        claim doesn't appear in any cited span (per-span check).
      - TITLE_MISMATCH: cited evidence's source title shares no
        content tokens with the claim (per-source check). 2026-05-02
        spin-glass case: claim about spin glass cited to *Quantum
        chromodynamics*.

    When both fire on different claims of the same answer, surface
    both tails so the operator sees the full picture."""
    violations = result.get("violations") or []
    kinds = {v.get("kind") for v in violations}
    parts: list[str] = []
    if "WARRANT_MISSING" in kinds:
        parts.append("warrant missing")
    if "TITLE_MISMATCH" in kinds:
        parts.append("title mismatch")
    if "FORMAT_COLLAPSED" in kinds:
        parts.append("format collapsed")
    # Ticket #000008 Phase 4 — broad-quantifier tails (§10.3 / §10.7).
    # Each names what the preflight detected so operators don't have to
    # parse violation lists by hand. Cap value comes from
    # `claim_cap_applied` on the result when present.
    if "BROAD_QUANTIFIER_REJECTED" in kinds:
        parts.append("broad rejected")
    elif "BROAD_QUANTIFIER_CAP_APPLIED" in kinds:
        cap = result.get("claim_cap_applied")
        parts.append(f"broad cap {cap}" if cap else "broad cap")
    elif "BROAD_QUANTIFIER_SCOPE_UNBOUND" in kinds:
        parts.append("broad unbounded")
    elif "BROAD_QUANTIFIER_RUNAWAY" in kinds:
        parts.append("broad runaway")
    # Ticket #000010 — meta-cognition logical-status tails. Pulled
    # from result["question_state"]["logical_statuses"] when present.
    # Doesn't double up with the broad-quantifier tails above (those
    # come from the verifier violation list, not the preflight).
    qs = result.get("question_state") or {}
    statuses = set(qs.get("logical_statuses") or [])
    if "false_premise_suspected" in statuses:
        parts.append("false premise")
    if "contradictory_question" in statuses:
        parts.append("contradictory")
    if "stale_risk" in statuses:
        parts.append("stale risk")
    if "out_of_corpus_risk" in statuses:
        parts.append("out of corpus")
    if "reference_frame_ambiguous" in statuses:
        parts.append("frame ambiguous")
    # Ticket #000011 — soft preflight sidecar hint. Renders distinctly
    # from the hard tails above so an operator can tell at a glance
    # that the signal is advisory. Skips SOFT_DISABLED / SOFT_PARSE_FAIL
    # / SOFT_WELL_FORMED (no actionable signal).
    soft = result.get("soft_preflight_hint") or {}
    soft_label = soft.get("classifier_label") or ""
    if soft_label and soft_label not in (
        "SOFT_DISABLED", "SOFT_PARSE_FAIL", "SOFT_WELL_FORMED",
    ):
        # Strip SOFT_ prefix + lowercase for tail readability
        # (e.g. SOFT_FALSE_PREMISE_SUSPECTED → "false premise suspected").
        readable = soft_label.removeprefix("SOFT_").lower().replace("_", " ")
        parts.append(f"soft: {readable}")
    if not parts:
        return ""
    return " · " + " · ".join(parts)


def _maybe_render_json_envelope_as_bullets(answer: str) -> str:
    """If `answer` is a claim_lattice JSON envelope, render bullets.

    JSON-mode runs that land UNGROUNDED have no verified claims so the
    runtime's bullet renderer produces empty text and `answer_text`
    falls back to the raw model output — a `{"claims":[...]}` envelope.
    The user then sees raw JSON for failed runs and bullets for
    successful ones, which reads as inconsistent. Detect the JSON
    shape, parse it (lenient), and render each claim's `text` as a
    bullet line tagged with its evidence_ids so the surface stays
    consistent across grounded / ungrounded outcomes.

    Falls back to the raw input unchanged if:
      - input doesn't look like JSON (no leading `{`)
      - parse fails (lenient parser exception)
      - parse succeeds but the shape isn't `{"claims": [...]}`
    """
    stripped = (answer or "").lstrip()
    if not stripped.startswith("{") and not stripped.startswith("```"):
        return answer
    if "claims" not in stripped:
        return answer
    try:
        from aborist.qa.verify import _lenient_json_parse
        parsed, _fixups = _lenient_json_parse(answer)
    except Exception:
        return answer
    if not isinstance(parsed, dict):
        return answer
    raw_claims = parsed.get("claims")
    if not isinstance(raw_claims, list) or not raw_claims:
        return answer
    out_lines: list[str] = []
    for c in raw_claims:
        if not isinstance(c, dict):
            continue
        text = c.get("text") or ""
        if not isinstance(text, str) or not text.strip():
            continue
        eids = c.get("evidence_ids") or []
        if isinstance(eids, list) and eids:
            ids = ",".join(str(x) for x in eids if isinstance(x, str))
            out_lines.append(f"- {text.strip()}  [{ids}: unverified]")
        else:
            out_lines.append(f"- {text.strip()}")
    return "\n".join(out_lines) if out_lines else answer


def _render_query_human(result: dict, question: str) -> str:
    """Pretty-print a query result for terminal reading.

    Layout:
        question
        AUDIT_MODE  N/M verified  via verifier_method  Xs  (cached|fresh)

        answer text...

        sources (K):
          [1] Title — host/path  (shard.db)
          [2] ...

        unverified (J):
          - "..."

        cache_key: 35ab7d33…   <run with --json for full record>

    Errors / no-source paths fall back to a short status line.
    """
    status = result.get("status")
    if status == "broad_quantifier_rejected":
        # Phase 4 reject-broad early-return path. The result carries
        # an answer_text with the rejection rationale + a violations
        # list; render both so the operator sees WHY without --json.
        answer_text = result.get("answer_text") or ""
        violations = result.get("violations") or []
        kind = next(
            (v.get("kind") for v in violations
             if v.get("kind") == "BROAD_QUANTIFIER_REJECTED"),
            "BROAD_QUANTIFIER_REJECTED",
        )
        intensity = result.get("quantifier_intensity") or "?"
        token = result.get("quantifier_matched_token") or "?"
        cap = result.get("claim_cap_applied")
        cap_str = f" · cap was {cap}" if cap else ""
        return (
            f"{question}\n"
            f"  UNGROUNDED · via {kind} · {intensity} (\"{token}\")"
            f"{cap_str}  0/0  0.0s  (preflight)\n\n"
            f"{answer_text}"
        )
    if status not in ("cache_hit", "cache_miss_then_written"):
        msg = result.get("msg") or status or "unknown error"
        return f"  {status or 'error'}: {msg}"

    audit = result.get("audit_mode", "?")
    n_quotes = result.get("n_quotes", 0) or 0
    n_verified = result.get("n_verified", 0) or 0
    method = result.get("verifier_method", "?")
    timings = result.get("timings") or {}
    total_ms = timings.get("total_ms")
    elapsed = f"{total_ms / 1000:.1f}s" if isinstance(total_ms, (int, float)) else "?"
    cache_status = "cached" if status == "cache_hit" else "fresh"
    lookup_path = result.get("lookup_path")
    # Annotate cache_hits when they came from a fallback ckey, not the
    # primary one — useful when an agent ran with fidelity=equivalence_class
    # and reused another agent's record.
    if lookup_path and lookup_path.endswith("_fallback"):
        cache_status = f"cached via {lookup_path}"

    # Render-layer label honesty: schema audit_mode (STRICT/HYBRID/
    # UNGROUNDED) describes what the lexical verifier checked, not
    # full semantic entailment. The display label combines audit_mode
    # with verifier_method so the user sees what was actually
    # verified. Four-rung ladder for claim-lattice methods (#000005):
    #   POINTER-LINKED      pointer verified; warrant didn't apply
    #                       or failed for some claim
    #   ANCHOR-WARRANTED    pointer-linked + warrant passed where
    #                       it ran; other soft demotes may apply
    #   EVIDENCE-WARRANTED  anchor-warranted + no soft demotes
    #   UNGROUNDED          no verified pairs
    # Schema column stays unchanged; pure display.
    display_label = _render_audit_label(audit, method, result.get("violations"))
    warrant_tail = _render_warrant_tail(result)

    lines: list[str] = []
    lines.append(question)
    lines.append(
        f"  {display_label}{warrant_tail}  {n_verified}/{n_quotes}  "
        f"{elapsed}  ({cache_status})"
    )
    lines.append("")

    answer = result.get("answer_text") or ""
    # When JSON-mode runs land UNGROUNDED, rendered_text is empty and
    # answer_text falls back to the raw model output — a JSON envelope.
    # Parse it and render each claim as a bullet so the user gets the
    # same shape whether the run grounded or not. Falls back to raw
    # display if parse fails or output isn't JSON-shaped.
    answer = _maybe_render_json_envelope_as_bullets(answer)
    lines.append(answer)
    lines.append("")

    sources = result.get("sources") or []
    if sources:
        lines.append(f"sources ({len(sources)}):")
        for i, s in enumerate(sources, start=1):
            uri = s.get("document_uri", "")
            title = (s.get("title") or "").strip() or _short_path(uri)
            shard = s.get("shard")
            shard_part = f"  ({shard})" if shard else ""
            # Render-layer source-role display + used/unused annotation
            # (claim_lattice modes only — quote-mode results don't carry
            # the per-source `used` flag). Honestly surfaces "the system
            # retrieved noise but did not rely on it" so the user can
            # see the model ignoring distractors instead of having to
            # infer it. Pre-2026-05-01 the source list showed every
            # retrieved doc indistinguishably.
            role = s.get("source_role")
            used = s.get("used")
            pointer_ids = s.get("used_pointer_ids") or []
            annotations: list[str] = []
            if role:
                annotations.append(role)
            if used is True:
                if pointer_ids:
                    annotations.append(f"used ({','.join(pointer_ids)})")
                else:
                    annotations.append("used")
            elif used is False:
                annotations.append("unused")
            if annotations:
                annotation_part = " — " + " — ".join(annotations)
            else:
                annotation_part = ""
            lines.append(
                f"  [{i}] {title}{annotation_part} — {_strip_scheme(uri)}{shard_part}"
            )
        lines.append("")

        # Retrieval-purity one-line summary (claim_lattice modes only).
        # "primary at #R · used N/M sources · M-N noise unused"
        # Surfaces noise-resistance at a glance without making the
        # user count rows themselves.
        purity = result.get("retrieval_purity")
        if purity:
            primary_rank = purity.get("primary_rank") or 0
            used = purity.get("used_sources", 0)
            total = purity.get("total_sources", 0)
            noise_unused = (
                purity.get("noise_sources_count", 0)
                - purity.get("noise_sources_used", 0)
            )
            primary_part = (
                f"primary at #{primary_rank}"
                if primary_rank > 0 else "no primary in top-K"
            )
            noise_part = (
                f" · {noise_unused} noise unused" if noise_unused else ""
            )
            lines.append(
                f"  retrieval purity: {primary_part} · used "
                f"{used}/{total} sources{noise_part}"
            )
            lines.append("")

    partially = result.get("partially_verified_quotes") or []
    if partially:
        lines.append(f"partially grounded ({len(partially)}):")
        for q in partially:
            qtxt = q if len(q) <= 100 else q[:97] + "..."
            lines.append(f'  - "{qtxt}"')
        lines.append("")

    unverified = result.get("unverified_quotes") or []
    if unverified:
        lines.append(f"unverified ({len(unverified)}):")
        for q in unverified:
            qtxt = q if len(q) <= 100 else q[:97] + "..."
            lines.append(f'  - "{qtxt}"')
        lines.append("")

    # Anchor-smell sidecar (claim_lattice mode only). Soft signal —
    # never persisted, never in cache_key. Surfaces when ≥50% of
    # verified claim-pointer pairs share one pointer_id AND there
    # are at least 3 verified pairs to compare; below 3 the ratio is
    # vacuous (1/1 always = 1.00 even when nothing is wrong).
    ratio = result.get("lazy_anchor_ratio")
    distribution = result.get("pointer_id_distribution") or {}
    total_pairs = sum(distribution.values()) if distribution else 0
    if (
        method == "claim_lattice"
        and isinstance(ratio, (int, float))
        and ratio >= 0.5
        and total_pairs >= 3
    ):
        top_pid, top_count = max(distribution.items(), key=lambda kv: kv[1])
        lines.append(
            f"lazy-anchor smell: {top_count} of {total_pairs} verified "
            f"pairs cite [{top_pid}] (ratio {ratio:.2f}); "
            f"distinct pointers cited: {len(distribution)}"
        )
        lines.append("")

    pc = result.get("prompt_chars") or {}
    if pc:
        # Compact one-liner — operator at a glance: did STRICT come
        # from a tight prompt or a context-stuffed one?
        lines.append(
            f"capacity: prompt {pc.get('messages_total', 0):,} chars "
            f"(sys {pc.get('system_prompt', 0):,} + "
            f"reminder {pc.get('grounding_reminder', 0):,} + "
            f"evidence {pc.get('evidence_or_context', 0):,} + "
            f"question {pc.get('user_question', 0):,}) → "
            f"answer {result.get('answer_chars', 0):,} chars"
        )

    # Per-phase timings — surfaces where the per-call cost lands.
    # Hermes-bound queries should show `llm_ms` dominating; if
    # search_ms or persist_ms creeps up that's a retrieval / WAL
    # signal the operator wants visible. Cache-hit rows skip llm
    # entirely so the breakdown also tells you which path you're
    # paying for.
    timings = result.get("timings") or {}
    if timings:
        parts: list[str] = []
        order = (
            ("cache_lookup_ms", "cache"),
            ("search_ms", "search"),
            ("context_ms", "context"),
            ("llm_ms", "llm"),
            ("persist_ms", "persist"),
        )
        for key, label in order:
            v = timings.get(key)
            if isinstance(v, (int, float)) and v > 0:
                parts.append(f"{label} {v / 1000:.2f}s")
        total = timings.get("total_ms")
        if isinstance(total, (int, float)):
            parts.append(f"**total {total / 1000:.2f}s**")
        if parts:
            lines.append("timings: " + " · ".join(parts))

    # Reference-frame notes (Ticket #000002). When detect_frame
    # classified the query as `reference`, surface the named work
    # so an operator knows the substrate routed to a fictional
    # source. Skipped for literal / no-phrase-route / ambiguous
    # rows — the line only appears when there's something to say.
    fd = result.get("frame_detection") or {}
    if fd.get("kind") == "reference" and fd.get("reference_title"):
        lines.append("")
        lines.append(f"reference frame: {fd['reference_title']}")
        if fd.get("reference_uri"):
            lines.append(f"  cited as the named work in the answer")

    cache_key = (result.get("cache_key") or "")[:8]
    lines.append(f"cache_key: {cache_key}…   <run with --json for full record>")
    return "\n".join(lines)


def _strip_scheme(uri: str) -> str:
    """`https://en.wikipedia.org/wiki/X` -> `en.wikipedia.org/wiki/X`."""
    for prefix in ("https://", "http://"):
        if uri.startswith(prefix):
            return uri[len(prefix):]
    return uri


def _short_path(uri: str) -> str:
    """Last URL segment as a fallback display name."""
    s = _strip_scheme(uri).rstrip("/")
    if "/" in s:
        return s.rsplit("/", 1)[1]
    return s


def _cmd_inspect(args: argparse.Namespace) -> int:
    """Sidecar diagnostic — pulls source chunks for a cache_key and
    classifies each unverified span. Read-only; no audit events, no
    providence_cache mutations.
    """
    from aborist.qa.inspect import inspect_cache_key

    qa_db = args.qa_db
    if qa_db is None:
        qa_db = (
            Path(args.global_shards_dir) / "qa.db"
            if args.global_shards_dir
            else Path.home() / ".aborist" / "qa.db"
        )
    shards_dir = (
        Path(args.global_shards_dir) if args.global_shards_dir else None
    )
    single_db = None if shards_dir else Path(args.db) if args.db else None

    result = inspect_cache_key(
        args.cache_key,
        qa_db=Path(qa_db),
        shards_dir=shards_dir,
        single_db=single_db,
    )

    if args.json:
        print(json.dumps(result, indent=2, ensure_ascii=False))
    else:
        print(_render_inspect_human(result))
    return 0 if result.get("status") == "ok" else 1


def _render_inspect_human(result: dict) -> str:
    """Pretty-print an inspect result so an operator can scan
    paraphrase vs invention vs trailing-artifact at a glance."""
    if result.get("status") != "ok":
        return f"  {result.get('status', 'error')}: cache_key={result.get('cache_key', '?')}"

    rec = result["record"]
    ctx = result["context"]
    sources = result["sources"]
    diagnoses = result["unverified"]

    lines: list[str] = []
    lines.append(rec["question_text"])
    lines.append(
        f"  {rec['audit_mode']}  {rec['n_verified']}/{rec['n_quotes']} verified  "
        f"via {rec['verifier_method']}  state={rec['falsification_state']}"
    )
    lines.append("")
    lines.append(
        f"context: {ctx['raw_chars']:,} raw -> {ctx['base_chars']:,} base "
        f"(wikitext-strip {'on' if ctx['wikitext_strip_active'] else 'off'})"
    )
    if sources:
        lines.append(f"sources ({len(sources)}):")
        for i, s in enumerate(sources, start=1):
            lines.append(
                f"  [{i}] {s.get('title') or '(untitled)'} — "
                f"{s.get('chunk_count', '?')} chunks, "
                f"{s.get('raw_chars', 0):,} chars"
            )
    lines.append("")

    if not diagnoses:
        lines.append("(no unverified spans)")
        return "\n".join(lines)

    lines.append(f"unverified diagnoses ({len(diagnoses)}):")
    for i, d in enumerate(diagnoses, start=1):
        span = d.get("span", "")
        diag = d.get("diagnosis", "?")
        lines.append("")
        lines.append(f"  [{i}] {diag}")
        lines.append(f"      span: {_short(span, 140)}")
        if diag == "trailing_artifact":
            lines.append(f"      matched_prefix_chars: {d.get('matched_prefix_chars')}")
            lines.append(f"      trailing_artifact: {_short(d.get('trailing_artifact', ''), 100)}")
        elif diag == "synthetic_elision_inside_quote":
            lines.append(
                f"      [...] inserted by model — "
                f"{d.get('prefix_chars', 0)} prefix chars "
                f"({'in source' if d.get('prefix_in_source') else 'NOT in source'}), "
                f"{d.get('suffix_chars', 0)} suffix chars "
                f"({'in source' if d.get('suffix_in_source') else 'NOT in source'})"
            )
        elif diag == "interior_elision":
            lines.append(
                f"      matched: {d.get('matched_prefix_chars')} prefix + "
                f"{d.get('matched_suffix_chars')} suffix chars (parenthetical aside dropped)"
            )
            lines.append(f"      dropped_aside: {_short(d.get('dropped_aside', ''), 120)}")
        elif diag in ("paraphrase", "partial_paraphrase"):
            lines.append(f"      token_coverage: {d.get('token_coverage')}")
            counts = d.get("token_counts", {})
            if counts:
                top = ", ".join(f"{k}×{v}" for k, v in list(counts.items())[:6])
                lines.append(f"      tokens_in_base: {top}")
            missing = d.get("missing_tokens", [])
            if missing:
                lines.append(f"      missing_tokens: {missing[:8]}")
        elif diag == "no_overlap":
            lines.append(f"      tokens_checked: {d.get('tokens_checked', '?')}")
            lines.append(f"      tokens_present: {d.get('tokens_present', '?')}")
        repair = d.get("repair")
        if repair:
            action = repair.get("action", "?")
            reason = repair.get("reason", "")
            line = f"      repair: {action}"
            if reason:
                line += f"  ({reason})"
            lines.append(line)
    return "\n".join(lines)


def _short(s: str, n: int) -> str:
    return s if len(s) <= n else s[: n - 3] + "..."


def _falsify_cache_key(
    cache_key_value: str,
    *,
    state: str,
    reason: str,
    by_actor: str,
    shards_dir: Path | None,
    db_path: Path | None,
) -> dict:
    """Mark a providence_cache record as failed/stale/quarantined across shards.

    Searches every shard for the cache_key (it lives in exactly one).
    Updates the row's falsification_state, appends a falsification log
    entry, and writes a 'falsify' audit event so the chain records the act.
    """
    import time as _time

    from aborist.store import append_audit, discover_shards, transaction

    if state not in ("failed", "stale", "quarantined"):
        return {"status": "invalid_state", "value": state}

    paths: list[Path] = (
        discover_shards(shards_dir) if shards_dir else [Path(db_path)]
    )

    for sp in paths:
        c = connect(sp)
        try:
            row = c.execute(
                "SELECT cache_key, falsification_state FROM providence_cache "
                "WHERE cache_key = ?",
                (cache_key_value,),
            ).fetchone()
            if row is None:
                continue
            now = int(_time.time())
            with transaction(c):
                event_hash = append_audit(
                    c,
                    event_type="falsify",
                    subject_root=cache_key_value,
                    body={
                        "cache_key": cache_key_value,
                        "from_state": row["falsification_state"],
                        "to_state": state,
                        "reason": reason,
                        "by_actor": by_actor,
                    },
                    ts=now,
                )
                c.execute(
                    "UPDATE providence_cache "
                    "SET falsification_state = ?, audit_event_hash = ? "
                    "WHERE cache_key = ?",
                    (state, event_hash, cache_key_value),
                )
                c.execute(
                    "INSERT INTO falsifications "
                    "(cache_key, state, reason, by_actor, at, audit_event_hash) "
                    "VALUES (?, ?, ?, ?, ?, ?)",
                    (cache_key_value, state, reason, by_actor, now, event_hash),
                )
            return {
                "status": "falsified",
                "cache_key": cache_key_value,
                "shard": sp.name,
                "from_state": row["falsification_state"],
                "to_state": state,
                "reason": reason,
                "by_actor": by_actor,
                "audit_event_hash": event_hash,
                "ts": now,
            }
        finally:
            c.close()

    return {"status": "not_found", "cache_key": cache_key_value}


def _burn_cache_key(
    cache_key_value: str,
    *,
    reason: str,
    by_actor: str,
    shards_dir: Path | None,
    db_path: Path | None,
    force: bool = False,
) -> dict:
    """Delete a providence_cache leaf, but only if it has no children.

    "Kindergarten of a tree's genesis" — early/scratch use. Falsify keeps
    history; burn removes the row. Children today = falsifications
    referencing this cache_key. If any exist, refuse without ``--force``.

    Always writes a 'providence_burn' audit event so the chain records
    that a leaf was removed and why. Use ``aborist providence --falsify``
    instead when downstream consumers may have built on this answer.
    """
    import time as _time

    from aborist.store import append_audit, discover_shards, transaction

    paths: list[Path] = (
        discover_shards(shards_dir) if shards_dir else [Path(db_path)]
    )

    for sp in paths:
        c = connect(sp)
        try:
            row = c.execute(
                "SELECT cache_key, audit_mode, n_verified, falsification_state, "
                " question_text FROM providence_cache WHERE cache_key = ?",
                (cache_key_value,),
            ).fetchone()
            if row is None:
                continue
            child_falsifications = c.execute(
                "SELECT COUNT(*) FROM falsifications WHERE cache_key = ?",
                (cache_key_value,),
            ).fetchone()[0]
            if child_falsifications > 0 and not force:
                return {
                    "status": "refused_has_children",
                    "cache_key": cache_key_value,
                    "shard": sp.name,
                    "child_falsifications": int(child_falsifications),
                    "hint": "use --force to burn anyway, or 'providence --falsify' to keep history",
                }
            now = int(_time.time())
            with transaction(c):
                c.execute(
                    "DELETE FROM providence_cache WHERE cache_key = ?",
                    (cache_key_value,),
                )
            event_hash = append_audit(
                c,
                event_type="providence_burn",
                subject_root=cache_key_value,
                body={
                    "cache_key": cache_key_value,
                    "burned_audit_mode": row["audit_mode"],
                    "burned_n_verified": int(row["n_verified"]),
                    "burned_state": row["falsification_state"],
                    "question_text": row["question_text"],
                    "reason": reason,
                    "by_actor": by_actor,
                    "child_falsifications_at_burn": int(child_falsifications),
                    "forced": bool(child_falsifications > 0 and force),
                },
                ts=now,
            )
            return {
                "status": "burned",
                "cache_key": cache_key_value,
                "shard": sp.name,
                "burned_audit_mode": row["audit_mode"],
                "reason": reason,
                "by_actor": by_actor,
                "audit_event_hash": event_hash,
                "ts": now,
            }
        finally:
            c.close()

    return {"status": "not_found", "cache_key": cache_key_value}


def _count_document_children(c, document_root: str) -> dict:
    """Count outbound child references that 'burn' must protect.

    For a document/core leaf, "children" = anything downstream that built on
    this row. Specifically:
      - derivations rows where ``src_root = root`` (a core was distilled
        from this — burning would orphan or silently cascade-truncate the
        derivation, leaving the descendant core dangling).
      - edges rows where ``dst_root = root`` (other documents link to this
        one; burning leaves dangling references).
      - providence_cache rows where ``source_root = root`` (Q&A grounded
        in this document).

    NOTE on schema: derivations has ON DELETE CASCADE on BOTH ``core_root``
    AND ``src_root``. Without this gate, a bare DELETE FROM documents would
    silently cascade-prune derivations and orphan downstream cores.
    """
    derivations_downstream = c.execute(
        "SELECT COUNT(*) FROM derivations WHERE src_root = ?",
        (document_root,),
    ).fetchone()[0]
    incoming_edges = c.execute(
        "SELECT COUNT(*) FROM edges WHERE dst_root = ?",
        (document_root,),
    ).fetchone()[0]
    providence_refs = c.execute(
        "SELECT COUNT(*) FROM providence_cache WHERE source_root = ?",
        (document_root,),
    ).fetchone()[0]
    return {
        "derivations_downstream": int(derivations_downstream),
        "incoming_edges": int(incoming_edges),
        "providence_refs": int(providence_refs),
    }


def _burn_document_root(
    document_root_value: str,
    *,
    reason: str,
    by_actor: str,
    shards_dir: Path | None,
    db_path: Path | None,
    force: bool = False,
) -> dict:
    """Delete a surface document leaf, but only if it has no children.

    Children:
      - derivations.src_root = root      (downstream cores derived from it)
      - edges.dst_root       = root      (other docs link to it)
      - providence_cache.source_root = root  (Q&A grounded in it)

    On burn:
      - DELETE FROM chunks_fts (FTS5 has no FK; clear before chunks vanish).
      - DELETE FROM documents — cascades to chunks + merkle_nodes via FK.
      - Append a 'document_burn' audit event recording counts + forced flag.

    Refuses with status='refused_has_children' (and skips the audit event)
    when any child count > 0 and ``--force`` is not set, so callers can fix
    state and retry idempotently.
    """
    import time as _time

    from aborist.store import append_audit, discover_shards, transaction

    paths: list[Path] = (
        discover_shards(shards_dir) if shards_dir else [Path(db_path)]
    )

    for sp in paths:
        c = connect(sp)
        try:
            row = c.execute(
                "SELECT document_root, document_uri, kind, title, source_type, "
                "       chunking_version, canonicalization_version, schema_version "
                "FROM documents WHERE document_root = ? AND kind = 'surface'",
                (document_root_value,),
            ).fetchone()
            if row is None:
                continue
            counts = _count_document_children(c, document_root_value)
            total_children = sum(counts.values())
            if total_children > 0 and not force:
                return {
                    "status": "refused_has_children",
                    "document_root": document_root_value,
                    "kind": "surface",
                    "shard": sp.name,
                    **counts,
                    "hint": "use --force to burn anyway (orphans descendants); "
                            "prefer evict for cold-tier compression",
                }
            chunk_count = c.execute(
                "SELECT COUNT(*) FROM chunks WHERE document_root = ?",
                (document_root_value,),
            ).fetchone()[0]
            now = int(_time.time())
            with transaction(c):
                # FTS5 has no FK to chunks; clear by chunk_id before the
                # CASCADE on documents wipes the rows that resolve them.
                for cr in c.execute(
                    "SELECT chunk_id FROM chunks WHERE document_root = ?",
                    (document_root_value,),
                ).fetchall():
                    c.execute(
                        "DELETE FROM chunks_fts WHERE rowid = ?",
                        (cr["chunk_id"],),
                    )
                # Outbound edges (src_root = this) carry no FK; clean them
                # explicitly so we don't leave half-edges pointing from a
                # ghost. Incoming edges (dst_root = this) are already gated
                # above by the children check.
                c.execute(
                    "DELETE FROM edges WHERE src_root = ?",
                    (document_root_value,),
                )
                # documents -> chunks/merkle_nodes/derivations cascade via FK.
                c.execute(
                    "DELETE FROM documents WHERE document_root = ?",
                    (document_root_value,),
                )
            event_hash = append_audit(
                c,
                event_type="document_burn",
                subject_root=document_root_value,
                body={
                    "document_root": document_root_value,
                    "document_uri": row["document_uri"],
                    "kind": "surface",
                    "title": row["title"],
                    "source_type": row["source_type"],
                    "burned_chunk_count": int(chunk_count),
                    "child_counts_at_burn": counts,
                    "reason": reason,
                    "by_actor": by_actor,
                    "forced": bool(total_children > 0 and force),
                },
                ts=now,
            )
            return {
                "status": "burned",
                "document_root": document_root_value,
                "kind": "surface",
                "shard": sp.name,
                "burned_chunk_count": int(chunk_count),
                "child_counts_at_burn": counts,
                "reason": reason,
                "by_actor": by_actor,
                "audit_event_hash": event_hash,
                "ts": now,
            }
        finally:
            c.close()

    return {"status": "not_found", "document_root": document_root_value, "kind": "surface"}


def _burn_core_root(
    document_root_value: str,
    *,
    reason: str,
    by_actor: str,
    shards_dir: Path | None,
    db_path: Path | None,
    force: bool = False,
) -> dict:
    """Delete a core document leaf, but only if it has no children.

    Same children gates as ``_burn_document_root`` (derivations.src_root,
    edges.dst_root, providence_cache.source_root). The "PLUS no further
    derivations build cores from this core" rule from the spec is
    structurally identical to derivations.src_root > 0 — a core acts as a
    src_root only when something deeper distilled from it.

    CLAUDE.md says "Cores never evict" — that's the eviction subsystem,
    which only touches kind='surface'. Burn is operator-driven removal:
    cores CAN be burned, but the children gate is enforced.

    Audit event type is 'core_burn' so chain consumers can distinguish
    surface vs core leaf removals at a glance.
    """
    import time as _time

    from aborist.store import append_audit, discover_shards, transaction

    paths: list[Path] = (
        discover_shards(shards_dir) if shards_dir else [Path(db_path)]
    )

    for sp in paths:
        c = connect(sp)
        try:
            row = c.execute(
                "SELECT document_root, document_uri, kind, title, source_type, "
                "       compression_depth, chunking_version, "
                "       canonicalization_version, schema_version "
                "FROM documents WHERE document_root = ? AND kind = 'core'",
                (document_root_value,),
            ).fetchone()
            if row is None:
                continue
            counts = _count_document_children(c, document_root_value)
            total_children = sum(counts.values())
            if total_children > 0 and not force:
                return {
                    "status": "refused_has_children",
                    "document_root": document_root_value,
                    "kind": "core",
                    "shard": sp.name,
                    **counts,
                    "hint": "use --force to burn anyway; cores carry "
                            "downstream derivations that will be orphaned",
                }
            chunk_count = c.execute(
                "SELECT COUNT(*) FROM chunks WHERE document_root = ?",
                (document_root_value,),
            ).fetchone()[0]
            # Inbound derivations (where this core is core_root, i.e. its
            # binding back to source surfaces). These are NOT children —
            # they're the core's own provenance and cascade-delete with it.
            inbound_derivations = c.execute(
                "SELECT COUNT(*) FROM derivations WHERE core_root = ?",
                (document_root_value,),
            ).fetchone()[0]
            now = int(_time.time())
            with transaction(c):
                for cr in c.execute(
                    "SELECT chunk_id FROM chunks WHERE document_root = ?",
                    (document_root_value,),
                ).fetchall():
                    c.execute(
                        "DELETE FROM chunks_fts WHERE rowid = ?",
                        (cr["chunk_id"],),
                    )
                c.execute(
                    "DELETE FROM edges WHERE src_root = ?",
                    (document_root_value,),
                )
                c.execute(
                    "DELETE FROM documents WHERE document_root = ?",
                    (document_root_value,),
                )
            event_hash = append_audit(
                c,
                event_type="core_burn",
                subject_root=document_root_value,
                body={
                    "document_root": document_root_value,
                    "document_uri": row["document_uri"],
                    "kind": "core",
                    "title": row["title"],
                    "source_type": row["source_type"],
                    "compression_depth": int(row["compression_depth"]),
                    "burned_chunk_count": int(chunk_count),
                    "burned_inbound_derivations": int(inbound_derivations),
                    "child_counts_at_burn": counts,
                    "reason": reason,
                    "by_actor": by_actor,
                    "forced": bool(total_children > 0 and force),
                },
                ts=now,
            )
            return {
                "status": "burned",
                "document_root": document_root_value,
                "kind": "core",
                "shard": sp.name,
                "burned_chunk_count": int(chunk_count),
                "burned_inbound_derivations": int(inbound_derivations),
                "child_counts_at_burn": counts,
                "reason": reason,
                "by_actor": by_actor,
                "audit_event_hash": event_hash,
                "ts": now,
            }
        finally:
            c.close()

    return {"status": "not_found", "document_root": document_root_value, "kind": "core"}


def _cmd_burn_kindergarten(args: argparse.Namespace) -> int:
    """Burn all providence_cache rows younger than the kindergarten window.

    Test-ergonomic mass burn: when iterating on retrieval/verifier knobs
    you want to wipe recent test runs without finding each cache_key.
    Mirrors the kindergarten window from `mesh sync` so what's still
    "private" (un-broadcast) is also what's safe to bust without
    confusing peers.

    Each row goes through the standard `_burn_cache_key` so the
    children gate is honored (use `--force` to override en masse).
    Each successful burn writes one ``providence_burn`` audit event;
    chain integrity is verifiable via `make chain-check-shards` after.
    """
    import time as _time
    from aborist.store import discover_shards

    now = int(_time.time())
    # `kindergarten_seconds <= 0` means "no time gate — burn every live
    # row" per the verb's docstring. The earlier `cutoff = now - 0`
    # treated 0 as a same-second-only window, which made
    # test_burn_kindergarten_zero_seconds_burns_everything timing-flaky:
    # if the wall-clock second rolled over between seed and burn,
    # cutoff > seed.created_at and nothing matched.
    kindergarten_seconds = max(0, args.kindergarten_seconds)
    no_time_gate = kindergarten_seconds == 0
    cutoff = 0 if no_time_gate else now - kindergarten_seconds
    shards_dir = Path(args.global_shards_dir) if args.global_shards_dir else None
    single_db = Path(args.db) if args.db else None
    paths: list[Path] = (
        discover_shards(shards_dir) if shards_dir else [single_db]
    )
    actor = args.by_actor or os.environ.get("USER", "unknown")
    reason = args.reason or f"burn-kindergarten window={args.kindergarten_seconds}s"

    examined = 0
    burned = 0
    refused = 0
    not_found = 0
    items: list[dict] = []
    for sp in paths:
        c = connect(sp)
        try:
            if no_time_gate:
                rows = c.execute(
                    "SELECT cache_key, created_at, audit_mode "
                    "FROM providence_cache "
                    "WHERE falsification_state = 'live' "
                    "ORDER BY created_at DESC"
                ).fetchall()
            else:
                rows = c.execute(
                    "SELECT cache_key, created_at, audit_mode "
                    "FROM providence_cache "
                    "WHERE created_at >= ? AND falsification_state = 'live' "
                    "ORDER BY created_at DESC",
                    (cutoff,),
                ).fetchall()
        finally:
            c.close()
        for r in rows:
            examined += 1
            if args.dry_run:
                items.append({
                    "cache_key": r["cache_key"],
                    "audit_mode": r["audit_mode"],
                    "created_at": r["created_at"],
                    "would_burn": True,
                })
                continue
            result = _burn_cache_key(
                r["cache_key"],
                reason=reason,
                by_actor=actor,
                shards_dir=shards_dir,
                db_path=single_db,
                force=bool(args.force),
            )
            status = result.get("status")
            if status == "burned":
                burned += 1
            elif status == "refused_has_children":
                refused += 1
            else:
                not_found += 1
            items.append(result)

    print(json.dumps({
        "status": "dry_run" if args.dry_run else "burned",
        "kindergarten_seconds": args.kindergarten_seconds,
        "cutoff_at": cutoff,
        "now": now,
        "examined": examined,
        "burned": burned,
        "refused_has_children": refused,
        "not_found": not_found,
        "items": items[: args.verbose],
    }, indent=2, ensure_ascii=False))
    return 0


def _cmd_burn(args: argparse.Namespace) -> int:
    """CLI: burn a leaf with no children.

    Dispatches on ``--kind`` to the matching helper. Default 'providence'
    preserves the original surface (`--cache-key` only) so existing scripts
    keep working. Document/core kinds use ``--root``.
    """
    kind = getattr(args, "kind", "providence") or "providence"
    shards = Path(args.global_shards_dir) if args.global_shards_dir else None
    db = Path(args.db) if args.db else None
    actor = args.by_actor or os.environ.get("USER", "unknown")
    reason = args.reason or ""
    force = bool(args.force)

    if kind == "providence":
        if not getattr(args, "cache_key", None):
            print("burn --kind providence requires --cache-key", file=sys.stderr)
            return 2
        result = _burn_cache_key(
            args.cache_key,
            reason=reason,
            by_actor=actor,
            shards_dir=shards,
            db_path=db,
            force=force,
        )
    elif kind in ("document", "core"):
        if not getattr(args, "root", None):
            print(f"burn --kind {kind} requires --root", file=sys.stderr)
            return 2
        helper = _burn_document_root if kind == "document" else _burn_core_root
        result = helper(
            args.root,
            reason=reason,
            by_actor=actor,
            shards_dir=shards,
            db_path=db,
            force=force,
        )
    else:
        print(f"unknown burn kind: {kind}", file=sys.stderr)
        return 2

    print(json.dumps(result, indent=2, ensure_ascii=False))
    return 0 if result.get("status") == "burned" else 1


def _cmd_providence(args: argparse.Namespace) -> int:
    """List providence_cache records or falsify one by cache_key."""
    if getattr(args, "falsify", None):
        result = _falsify_cache_key(
            args.falsify,
            state=args.state,
            reason=args.reason or "",
            by_actor=args.by_actor or os.environ.get("USER", "unknown"),
            shards_dir=Path(args.global_shards_dir) if args.global_shards_dir else None,
            db_path=Path(args.db) if args.db else None,
        )
        print(json.dumps(result, indent=2, ensure_ascii=False))
        return 0 if result.get("status") == "falsified" else 1
    if getattr(args, "show_preflight", None):
        # Ticket #000009 §7.2 — pull the preflight stage payload
        # from a row's run_dag_blob. Operator tool for inspecting
        # the policy state that governed the cached row.
        return _cmd_providence_show_preflight(
            cache_key_prefix=args.show_preflight,
            shards_dir=args.global_shards_dir,
            db=args.db,
        )
    conn = (
        connect_query(args.db, shards_dir=args.global_shards_dir)
        if args.global_shards_dir
        else connect(args.db)
    )
    try:
        if args.document_uri:
            rows = conn.execute(
                "SELECT cache_key, question_text, answer_text, falsification_state, "
                " hit_count, created_at FROM providence_cache "
                "WHERE document_uri = ? ORDER BY created_at DESC",
                (args.document_uri,),
            ).fetchall()
        elif args.source_root:
            rows = conn.execute(
                "SELECT cache_key, question_text, answer_text, falsification_state, "
                " hit_count, created_at FROM providence_cache "
                "WHERE source_root = ? ORDER BY created_at DESC",
                (args.source_root,),
            ).fetchall()
        else:
            rows = conn.execute(
                "SELECT cache_key, question_text, answer_text, falsification_state, "
                " hit_count, created_at FROM providence_cache "
                "ORDER BY created_at DESC LIMIT ?",
                (args.limit,),
            ).fetchall()
    finally:
        conn.close()
    out = [
        {
            "cache_key": r["cache_key"],
            "state": r["falsification_state"],
            "hit_count": r["hit_count"],
            "question": r["question_text"],
            "answer": r["answer_text"],
            "created_at": r["created_at"],
        }
        for r in rows
    ]
    print(json.dumps(out, indent=2, ensure_ascii=False))
    return 0


def _cmd_providence_show_preflight(
    *,
    cache_key_prefix: str,
    shards_dir: str | None,
    db: str | None,
) -> int:
    """Render the ``preflight`` stage payload for a cached row.

    Ticket #000009 §7.2. Pulls ``run_dag_blob`` for the matching
    cache row, parses the JSON, finds the ``preflight`` node, and
    pretty-prints the five nested CTI clauses (classifier,
    answer_contract, prompt_contract, evidence_contract,
    policy_refs) plus the metacognition QuestionState.

    Match is by 12-char prefix on ``cache_key`` (matches what
    bench rows + `_render_query_human` already truncate to).
    Returns 0 on success, 1 on miss / parse failure.
    """
    conn = (
        connect_query(db, shards_dir=shards_dir)
        if shards_dir
        else connect(db)
    )
    try:
        rows = conn.execute(
            "SELECT cache_key, question_text, run_dag_blob "
            "FROM providence_cache WHERE cache_key LIKE ? "
            "ORDER BY created_at DESC LIMIT 5",
            (cache_key_prefix + "%",),
        ).fetchall()
    finally:
        conn.close()
    if not rows:
        print(
            f"  no providence_cache row matching cache_key prefix "
            f"'{cache_key_prefix}'", file=sys.stderr,
        )
        return 1
    if len(rows) > 1:
        print(
            f"  {len(rows)} rows match prefix '{cache_key_prefix}'; "
            "rendering most recent. Pass a longer prefix to disambiguate.",
            file=sys.stderr,
        )
    row = rows[0]
    blob = row["run_dag_blob"]
    if not blob:
        print(
            f"  cache_key {row['cache_key'][:12]}: no run_dag_blob "
            "(legacy row, predates #000009)",
            file=sys.stderr,
        )
        return 1
    try:
        parsed = json.loads(blob)
    except json.JSONDecodeError as exc:
        print(f"  run_dag_blob parse error: {exc}", file=sys.stderr)
        return 1
    nodes = parsed.get("nodes") or []
    preflight_node = next(
        (n for n in nodes if isinstance(n, dict)
         and n.get("stage") == "preflight"),
        None,
    )
    if preflight_node is None:
        print(
            f"  cache_key {row['cache_key'][:12]}: run_dag has no "
            "preflight stage (predates #000009 binding)",
            file=sys.stderr,
        )
        return 1
    # Pull the full preflight payload (Ticket #000009 §7.2 — payload
    # now persisted alongside nodes via build_run_dag's
    # preflight_payload kwarg). Fall back to hash-only render for
    # legacy rows whose blob predates the payload-storage commit.
    payload = parsed.get("preflight_payload")
    out: dict = {
        "cache_key": row["cache_key"][:12],
        "question": row["question_text"],
        "preflight_stage_hash": preflight_node.get("hash"),
        "preflight_hash_12": (preflight_node.get("hash") or "")[:12],
        "run_dag_root": parsed.get("root"),
        "run_dag_stages": [n.get("stage") for n in nodes],
    }
    if payload is not None:
        # Verify the persisted payload hashes to the persisted leaf.
        # Mismatch would indicate post-write tampering or a serialization
        # drift; surface it explicitly so an auditor can detect.
        from aborist.qa.dag import _canonical_json, _sha256_hex
        recomputed = _sha256_hex(_canonical_json(payload))
        out["preflight_payload"] = payload
        out["payload_hash_check"] = (
            "ok" if recomputed == preflight_node.get("hash")
            else f"MISMATCH (recomputed {recomputed[:12]} != stored {preflight_node.get('hash', '')[:12]})"
        )
    else:
        out["preflight_payload"] = None
        out["payload_hash_check"] = (
            "unavailable: legacy row predates preflight_payload "
            "persistence (Ticket #000009 §7.2)"
        )
    print(json.dumps(out, indent=2, ensure_ascii=False))
    return 0


def _load_record_context(row, shards_dir, qa_db):
    """Reassemble context for a providence record. Returns text or None
    if any source doc has no hot chunks (cold)."""
    from aborist.qa.query import _load_doc_text

    proof = json.loads(row["merkle_proof"])
    sources = proof.get("sources", [])
    if not sources:
        return None
    parts: list[str] = []
    for src in sources:
        shard_name = src.get("shard")
        if not shard_name:
            return None
        if shards_dir:
            shard_path = shards_dir / shard_name
        else:
            shard_path = qa_db.parent / shard_name
        if not shard_path.exists():
            return None
        text = _load_doc_text(str(shard_path), src["document_root"])
        if not text:
            return None
        parts.append(text)
    return "\n\n".join(parts)


def _cmd_reclassify(args: argparse.Namespace) -> int:
    """Re-run the layered verifier against existing live providence records.

    Reads each record's answer + reassembles its context from
    merkle_proof.sources, runs verify_quotes(), and updates the row only
    if the verdict differs from what's stored. No LLM calls — this just
    relabels existing answers under the current verifier.

    Cold-source records (where any source doc has no hot chunks) are
    skipped: we can't faithfully reclassify without the original context.
    Run `aborist rehydrate` first if you want those covered too.

    `--compare` runs all four entity policies side-by-side without
    writing — use it to see what each policy would produce on real data
    before committing to one. `--entity-policy X` writes under a single
    policy.

    Each changed record gets one 'providence_reclassify' audit event
    with old & new state for chain-of-custody.
    """
    import time
    from collections import defaultdict

    from aborist.qa.verify import (
        DEFAULT_ENTITY_POLICY,
        ENTITY_POLICIES,
        verify_quotes,
    )

    qa_db = args.qa_db
    if qa_db is None:
        if args.global_shards_dir:
            qa_db = Path(args.global_shards_dir) / "qa.db"
        else:
            qa_db = Path.home() / ".aborist" / "qa.db"
    qa_db = Path(qa_db)

    shards_dir = (
        Path(args.global_shards_dir) if args.global_shards_dir else None
    )

    conn = connect(qa_db)
    try:
        sql = (
            "SELECT cache_key, answer_text, merkle_proof, audit_mode, "
            " verifier_method, n_quotes, n_verified, unverified_quotes, "
            " question_text "
            "FROM providence_cache "
            "WHERE falsification_state = 'live' "
            "ORDER BY created_at DESC"
        )
        if args.limit:
            sql += f" LIMIT {int(args.limit)}"
        rows = conn.execute(sql).fetchall()

        if args.compare:
            # Run all four policies side-by-side, no DB write. Output is a
            # per-record grid + a per-policy distribution summary so fox can
            # eyeball where the policies disagree.
            grid = []
            distribution: dict[str, dict[str, int]] = {
                p: defaultdict(int) for p in ENTITY_POLICIES
            }
            skipped_cold = 0
            for row in rows:
                context = _load_record_context(row, shards_dir, qa_db)
                if context is None:
                    skipped_cold += 1
                    continue
                per_policy = {}
                for p in ENTITY_POLICIES:
                    v = verify_quotes(row["answer_text"], context, entity_policy=p)
                    label = f"{v['audit_mode']}/{v['verifier_method']}"
                    per_policy[p] = label
                    distribution[p][label] += 1
                grid.append({
                    "cache_key": row["cache_key"][:16] + "…",
                    "question": row["question_text"][:55],
                    **per_policy,
                })
            print(json.dumps({
                "examined": len(grid),
                "skipped_cold": skipped_cold,
                "distribution": {p: dict(d) for p, d in distribution.items()},
                "records": grid,
            }, indent=2, ensure_ascii=False))
            return 0

        # Single-policy reclassify. Default tracks DEFAULT_ENTITY_POLICY
        # so the CLI always matches the verifier's current contract.
        policy_name = args.entity_policy or DEFAULT_ENTITY_POLICY
        if policy_name not in ENTITY_POLICIES:
            print(
                f"--entity-policy must be one of {ENTITY_POLICIES}, "
                f"got {policy_name!r}",
                file=sys.stderr,
            )
            return 2

        summary = {
            "examined": 0,
            "changed": 0,
            "skipped_cold": 0,
            "unchanged": 0,
            "entity_policy": policy_name,
            "transitions": defaultdict(int),
        }

        for row in rows:
            summary["examined"] += 1
            context = _load_record_context(row, shards_dir, qa_db)
            if context is None:
                summary["skipped_cold"] += 1
                continue

            verdict = verify_quotes(
                row["answer_text"], context, entity_policy=policy_name
            )

            old_unverified = row["unverified_quotes"] or "null"
            new_unverified_blob = (
                json.dumps(verdict["unverified_quotes"], separators=(",", ":"))
                if verdict["unverified_quotes"]
                else None
            )
            new_unverified_for_compare = new_unverified_blob or "null"

            unchanged = (
                verdict["audit_mode"] == row["audit_mode"]
                and verdict["verifier_method"] == row["verifier_method"]
                and verdict["n_quotes"] == row["n_quotes"]
                and verdict["n_verified"] == row["n_verified"]
                and old_unverified == new_unverified_for_compare
            )
            if unchanged:
                summary["unchanged"] += 1
                continue

            summary["changed"] += 1
            transition = (
                f"{row['audit_mode']}/{row['verifier_method']} "
                f"-> {verdict['audit_mode']}/{verdict['verifier_method']}"
            )
            summary["transitions"][transition] += 1

            if args.dry_run:
                continue

            now = int(time.time())
            with transaction(conn):
                event_hash = append_audit(
                    conn,
                    event_type="providence_reclassify",
                    subject_root=row["cache_key"],
                    body={
                        "old_audit_mode": row["audit_mode"],
                        "new_audit_mode": verdict["audit_mode"],
                        "old_method": row["verifier_method"],
                        "new_method": verdict["verifier_method"],
                        "old_n_verified": row["n_verified"],
                        "new_n_verified": verdict["n_verified"],
                        "entity_policy": policy_name,
                    },
                    ts=now,
                )
                conn.execute(
                    "UPDATE providence_cache SET "
                    " audit_mode = ?, n_quotes = ?, n_verified = ?, "
                    " unverified_quotes = ?, verifier_method = ?, "
                    " audit_event_hash = ? "
                    "WHERE cache_key = ?",
                    (
                        verdict["audit_mode"],
                        verdict["n_quotes"],
                        verdict["n_verified"],
                        new_unverified_blob,
                        verdict["verifier_method"],
                        event_hash,
                        row["cache_key"],
                    ),
                )
    finally:
        conn.close()

    summary["transitions"] = dict(summary["transitions"])
    summary["dry_run"] = bool(args.dry_run)
    print(json.dumps(summary, indent=2, ensure_ascii=False))
    return 0


def _cmd_emergent(args: argparse.Namespace) -> int:
    """Surface emergent claims from UNGROUNDED/HYBRID providence records.

    These are spans the model produced that don't appear verbatim in the
    corpus — candidate ingest targets. Frequent unverified quotes signal
    knowledge the model has from training that our corpus is missing.
    """
    conn = (
        connect_query(args.db, shards_dir=args.global_shards_dir)
        if args.global_shards_dir
        else connect(args.db)
    )
    try:
        if args.aggregate:
            rows = conn.execute(
                "SELECT unverified_quotes FROM providence_cache "
                "WHERE audit_mode IN ('UNGROUNDED','HYBRID') "
                " AND falsification_state = 'live' "
                " AND unverified_quotes IS NOT NULL"
            ).fetchall()
            counts: dict[str, int] = {}
            for r in rows:
                for q in json.loads(r["unverified_quotes"]):
                    counts[q] = counts.get(q, 0) + 1
            ranked = sorted(counts.items(), key=lambda kv: -kv[1])[: args.limit]
            print(json.dumps(
                [{"quote": q, "count": c} for q, c in ranked],
                indent=2, ensure_ascii=False
            ))
        else:
            rows = conn.execute(
                "SELECT cache_key, audit_mode, verifier_method, question_text, "
                " n_quotes, n_verified, unverified_quotes, created_at "
                "FROM providence_cache "
                "WHERE audit_mode IN ('UNGROUNDED','HYBRID') "
                " AND falsification_state = 'live' "
                "ORDER BY created_at DESC LIMIT ?",
                (args.limit,),
            ).fetchall()
            out = [
                {
                    "cache_key": r["cache_key"],
                    "audit_mode": r["audit_mode"],
                    "verifier_method": r["verifier_method"],
                    "question": r["question_text"],
                    "n_quotes": r["n_quotes"],
                    "n_verified": r["n_verified"],
                    "unverified_quotes": (
                        json.loads(r["unverified_quotes"])
                        if r["unverified_quotes"]
                        else []
                    ),
                    "created_at": r["created_at"],
                }
                for r in rows
            ]
            print(json.dumps(out, indent=2, ensure_ascii=False))
    finally:
        conn.close()
    return 0


def _cmd_evict(args: argparse.Namespace) -> int:
    from aborist.evict import evict_to_cold

    conn = (
        connect_query(args.db, shards_dir=args.global_shards_dir)
        if args.global_shards_dir
        else connect(args.db)
    )
    try:
        result = evict_to_cold(
            conn,
            source_type=args.source_type,
            older_than_days=args.older_than_days,
            document_roots=args.document_root or None,
        )
    finally:
        conn.close()
    print(json.dumps(result, indent=2, ensure_ascii=False))
    return 0


def _cmd_rehydrate(args: argparse.Namespace) -> int:
    from aborist.evict import rehydrate

    conn = (
        connect_query(args.db, shards_dir=args.global_shards_dir)
        if args.global_shards_dir
        else connect(args.db)
    )
    try:
        if args.all_cold:
            roots = [
                r["document_root"]
                for r in conn.execute(
                    "SELECT DISTINCT document_root FROM chunks WHERE tier = 'cold'"
                ).fetchall()
            ]
        else:
            roots = list(args.document_root or [])
        if not roots:
            print(
                "rehydrate needs --document-root R or --all-cold",
                file=sys.stderr,
            )
            return 2
        results = []
        for r in roots:
            res = rehydrate(conn, r)
            res["document_root"] = r
            results.append(res)
    finally:
        conn.close()
    print(json.dumps(results, indent=2, ensure_ascii=False))
    drift = sum(1 for r in results if r.get("status") == "drift_detected")
    return 1 if drift else 0


def _cmd_activity(args: argparse.Namespace) -> int:
    """Recent activity: Q&A records + freshly cached docs across all shards.

    Designed for an agent to read before deciding the next action — what was
    just asked, what was just integrated, what's the corpus state.
    """
    import time as _time

    from aborist.store import discover_shards

    shard_paths: list[Path] = []
    if args.global_shards_dir:
        shard_paths = discover_shards(args.global_shards_dir)
    else:
        shard_paths = [Path(args.db)]

    cutoff_ts = 0
    if args.since_seconds:
        cutoff_ts = int(_time.time()) - args.since_seconds

    qa_records: list[dict] = []
    ingest_events: list[dict] = []
    derive_events: list[dict] = []
    falsifications: list[dict] = []
    corpus = {
        "documents_total": 0,
        "documents_surface": 0,
        "documents_core": 0,
        "providence_total": 0,
        "providence_live": 0,
        "providence_stale": 0,
        "providence_failed": 0,
        "audit_events_total": 0,
    }

    for sp in shard_paths:
        c = connect(sp)
        try:
            corpus["documents_total"] += c.execute(
                "SELECT COUNT(*) FROM documents"
            ).fetchone()[0]
            corpus["documents_surface"] += c.execute(
                "SELECT COUNT(*) FROM documents WHERE kind='surface'"
            ).fetchone()[0]
            corpus["documents_core"] += c.execute(
                "SELECT COUNT(*) FROM documents WHERE kind='core'"
            ).fetchone()[0]
            corpus["providence_total"] += c.execute(
                "SELECT COUNT(*) FROM providence_cache"
            ).fetchone()[0]
            corpus["providence_live"] += c.execute(
                "SELECT COUNT(*) FROM providence_cache WHERE falsification_state='live'"
            ).fetchone()[0]
            corpus["providence_stale"] += c.execute(
                "SELECT COUNT(*) FROM providence_cache WHERE falsification_state='stale'"
            ).fetchone()[0]
            corpus["providence_failed"] += c.execute(
                "SELECT COUNT(*) FROM providence_cache WHERE falsification_state='failed'"
            ).fetchone()[0]
            corpus["audit_events_total"] += c.execute(
                "SELECT COUNT(*) FROM audit_events"
            ).fetchone()[0]

            # Q&A records
            for r in c.execute(
                "SELECT cache_key, question_text, answer_text, "
                " falsification_state, hit_count, created_at, last_hit_at, "
                " document_uri FROM providence_cache "
                "WHERE created_at >= ? ORDER BY created_at DESC LIMIT ?",
                (cutoff_ts, args.limit),
            ).fetchall():
                ans = r["answer_text"] or ""
                qa_records.append(
                    {
                        "ts": r["created_at"],
                        "shard": sp.name,
                        "cache_key": r["cache_key"],
                        "question": r["question_text"],
                        "answer_preview": (
                            ans if len(ans) <= args.preview_chars
                            else ans[: args.preview_chars] + "…"
                        ),
                        "sources_uri": r["document_uri"],
                        "state": r["falsification_state"],
                        "hit_count": r["hit_count"],
                        "last_hit_at": r["last_hit_at"],
                    }
                )

            # Recent ingest events
            for r in c.execute(
                "SELECT subject_root, body, ts FROM audit_events "
                "WHERE event_type = 'ingest' AND ts >= ? "
                "ORDER BY ts DESC LIMIT ?",
                (cutoff_ts, args.limit),
            ).fetchall():
                body = json.loads(r["body"]) if r["body"] else {}
                ingest_events.append(
                    {
                        "ts": r["ts"],
                        "shard": sp.name,
                        "document_root": r["subject_root"],
                        "document_uri": body.get("document_uri"),
                        "source_type": body.get("source_type"),
                        "chunks": body.get("chunks"),
                        "supersedes": body.get("supersedes"),
                    }
                )

            # Recent derive events (distillations)
            for r in c.execute(
                "SELECT subject_root, body, ts FROM audit_events "
                "WHERE event_type = 'derive' AND ts >= ? "
                "ORDER BY ts DESC LIMIT ?",
                (cutoff_ts, args.limit),
            ).fetchall():
                body = json.loads(r["body"]) if r["body"] else {}
                derive_events.append(
                    {
                        "ts": r["ts"],
                        "shard": sp.name,
                        "core_root": r["subject_root"],
                        "src_root": body.get("src_root"),
                        "process_id": body.get("process_id"),
                        "compression_ratio": body.get("compression_ratio"),
                        "compression_depth": body.get("compression_depth"),
                    }
                )

            # Recent falsifications
            for r in c.execute(
                "SELECT cache_key, state, reason, by_actor, at FROM falsifications "
                "WHERE at >= ? ORDER BY at DESC LIMIT ?",
                (cutoff_ts, args.limit),
            ).fetchall():
                falsifications.append(
                    {
                        "ts": r["at"],
                        "shard": sp.name,
                        "cache_key": r["cache_key"],
                        "state": r["state"],
                        "reason": r["reason"],
                        "by_actor": r["by_actor"],
                    }
                )
        finally:
            c.close()

    qa_records.sort(key=lambda x: -x["ts"])
    ingest_events.sort(key=lambda x: -x["ts"])
    derive_events.sort(key=lambda x: -x["ts"])
    falsifications.sort(key=lambda x: -x["ts"])

    print(
        json.dumps(
            {
                "as_of": int(_time.time()),
                "shards": [str(p) for p in shard_paths],
                "corpus": corpus,
                "recent_qa": qa_records[: args.limit],
                "recent_ingests": ingest_events[: args.limit],
                "recent_derives": derive_events[: args.limit],
                "recent_falsifications": falsifications[: args.limit],
            },
            indent=2, ensure_ascii=False
        )
    )
    return 0


def _cmd_stats(args: argparse.Namespace) -> int:
    conn = (
        connect_query(args.db, shards_dir=args.global_shards_dir)
        if args.global_shards_dir
        else connect(args.db)
    )
    try:
        result = stats(conn)
    finally:
        conn.close()
    print(json.dumps(result, indent=2, ensure_ascii=False))
    return 0


def _check_audit_chain(conn: sqlite3.Connection) -> tuple[int, int]:
    """Return (events_checked, breaks) for one audit chain in `conn`."""
    import hashlib

    rows = conn.execute(
        "SELECT seq, event_hash, prev_event_hash, body FROM audit_events ORDER BY seq"
    ).fetchall()
    prev = None
    breaks = 0
    for r in rows:
        h = hashlib.sha256()
        if r["prev_event_hash"]:
            h.update(bytes.fromhex(r["prev_event_hash"]))
        h.update(r["body"].encode("utf-8"))
        if h.hexdigest() != r["event_hash"]:
            breaks += 1
        if r["prev_event_hash"] != prev:
            breaks += 1
        prev = r["event_hash"]
    return len(rows), breaks


def _cmd_analyze(args: argparse.Namespace) -> int:
    """Compression spectrum, depth distribution, audit chain integrity."""
    from aborist.store import discover_shards

    # In sharded mode, audit chains live per-shard (each shard has its own
    # genesis -> latest). Check each independently and aggregate.
    audit_summary: dict | None = None
    if args.global_shards_dir:
        per_shard_chain = []
        total_events = 0
        total_breaks = 0
        for sp in discover_shards(args.global_shards_dir):
            sc = connect(sp)
            try:
                ev, br = _check_audit_chain(sc)
            finally:
                sc.close()
            per_shard_chain.append({"shard": sp.name, "events": ev, "breaks": br})
            total_events += ev
            total_breaks += br
        audit_summary = {
            "events": total_events,
            "breaks": total_breaks,
            "shards": per_shard_chain,
        }

    conn = (
        connect_query(args.db, shards_dir=args.global_shards_dir)
        if args.global_shards_dir
        else connect(args.db)
    )
    try:
        # Depth distribution.
        depth = conn.execute(
            "SELECT compression_depth, COUNT(*) AS n "
            "FROM documents GROUP BY compression_depth ORDER BY 1"
        ).fetchall()

        # Per-process compression ratios.
        procs = conn.execute(
            "SELECT json_extract(body, '$.process_id') AS process_id, "
            "       json_extract(body, '$.src_kind') AS src_kind, "
            "       AVG(CAST(json_extract(body, '$.compression_ratio') AS REAL)) AS mean_ratio, "
            "       MIN(CAST(json_extract(body, '$.compression_ratio') AS REAL)) AS min_ratio, "
            "       MAX(CAST(json_extract(body, '$.compression_ratio') AS REAL)) AS max_ratio, "
            "       COUNT(*) AS n_events "
            "FROM audit_events WHERE event_type='derive' "
            "GROUP BY process_id, src_kind"
        ).fetchall()

        # Source/kind crosstab.
        kinds = conn.execute(
            "SELECT source_type, kind, COUNT(*) AS n "
            "FROM documents GROUP BY source_type, kind ORDER BY 3 DESC"
        ).fetchall()

        # Tier distribution.
        tiers = conn.execute(
            "SELECT tier, COUNT(*) AS n FROM chunks GROUP BY tier"
        ).fetchall()

        # Top inbound link targets (the 'gravity wells' of the corpus).
        gravity = conn.execute(
            "SELECT dst_uri, COUNT(*) AS inbound FROM edges "
            "WHERE edge_type='wikilink' AND dst_uri != '' "
            "GROUP BY dst_uri ORDER BY inbound DESC LIMIT ?",
            (args.gravity_top,),
        ).fetchall()

        # Audit chain integrity (per-shard if sharded; single chain otherwise).
        if audit_summary is None:
            ev, br = _check_audit_chain(conn)
            audit_summary = {"events": ev, "breaks": br}

        report = {
            "compression_depth_histogram": [
                {"depth": r["compression_depth"], "count": r["n"]} for r in depth
            ],
            "distillers": [
                {
                    "process_id": r["process_id"],
                    "src_kind": r["src_kind"],
                    "n_events": r["n_events"],
                    "compression_ratio": {
                        "mean": (
                            round(r["mean_ratio"], 4)
                            if r["mean_ratio"] is not None
                            else None
                        ),
                        "min": (
                            round(r["min_ratio"], 4)
                            if r["min_ratio"] is not None
                            else None
                        ),
                        "max": (
                            round(r["max_ratio"], 4)
                            if r["max_ratio"] is not None
                            else None
                        ),
                    },
                }
                for r in procs
            ],
            "documents_by_source_kind": [
                {"source_type": r["source_type"], "kind": r["kind"], "count": r["n"]}
                for r in kinds
            ],
            "chunks_by_tier": {r["tier"]: r["n"] for r in tiers},
            "audit_chain": audit_summary,
            "gravity_top_inbound": [
                {"uri": r["dst_uri"], "inbound": r["inbound"]} for r in gravity
            ],
        }
    finally:
        conn.close()
    print(json.dumps(report, indent=2, ensure_ascii=False))
    return 0


def _cmd_snapshot_create(args: argparse.Namespace) -> int:
    """Compute snapshot_root over the read scope, persist into args.db.

    Single-DB mode (--db only): read + write are the same connection;
    delegate to the snapshot module's create_snapshot().

    Cross-shard mode (--shards-dir + --db): read against the in-memory
    UNION view to get the cluster-level Merkle root, then persist into
    args.db (a dedicated snapshots store, conventionally
    `~/.aborist/shards/snapshots.db`). The writer's own documents table
    is irrelevant to the snapshot value — only the union scope counts.
    """
    import time as _time

    from aborist.snapshot import compute_snapshot_root, create_snapshot

    if args.global_shards_dir is None:
        conn = connect(args.db)
        try:
            result = create_snapshot(
                conn, reason=args.reason, parent_snapshot=args.parent,
            )
        finally:
            conn.close()
        print(json.dumps(result, indent=2, ensure_ascii=False))
        return 0

    # Cross-shard: compute against UNION, write to args.db.
    read_conn = connect_query(args.db, shards_dir=args.global_shards_dir)
    try:
        snapshot_root, doc_count = compute_snapshot_root(read_conn)
    finally:
        read_conn.close()

    write_conn = connect(args.db)
    try:
        parent = args.parent
        if parent is None:
            row = write_conn.execute(
                "SELECT snapshot_root FROM snapshots ORDER BY taken_at DESC LIMIT 1"
            ).fetchone()
            if row is not None:
                parent = row["snapshot_root"]

        now = int(_time.time())
        body = {
            "snapshot_root": snapshot_root,
            "doc_count": doc_count,
            "parent_snapshot": parent,
            "reason": args.reason,
            "scope": "shards-union",
        }
        audit_event_hash = append_audit(
            write_conn,
            event_type="snapshot_create",
            body=body,
            subject_root=snapshot_root,
            ts=now,
        )
        with transaction(write_conn):
            write_conn.execute(
                "INSERT OR IGNORE INTO snapshots "
                "(snapshot_root, taken_at, audit_event_hash, doc_count, "
                " parent_snapshot, reason) VALUES (?, ?, ?, ?, ?, ?)",
                (
                    snapshot_root,
                    now,
                    audit_event_hash,
                    doc_count,
                    parent,
                    args.reason,
                ),
            )
    finally:
        write_conn.close()

    print(
        json.dumps(
            {
                "snapshot_root": snapshot_root,
                "doc_count": doc_count,
                "parent_snapshot": parent,
                "audit_event_hash": audit_event_hash,
                "taken_at": now,
                "reason": args.reason,
                "scope": "shards-union",
            },
            indent=2, ensure_ascii=False
        )
    )
    return 0


def _cmd_snapshot_list(args: argparse.Namespace) -> int:
    from aborist.snapshot import list_snapshots

    conn = connect(args.db)
    try:
        rows = list_snapshots(conn, limit=args.limit)
    finally:
        conn.close()
    print(json.dumps(rows, indent=2, ensure_ascii=False))
    return 0


def _cmd_snapshot_verify(args: argparse.Namespace) -> int:
    from aborist.snapshot import verify_snapshot

    conn = (
        connect_query(args.db, shards_dir=args.global_shards_dir)
        if args.global_shards_dir
        else connect(args.db)
    )
    try:
        result = verify_snapshot(conn, args.snapshot_root)
    finally:
        conn.close()
    print(json.dumps(result, indent=2, ensure_ascii=False))
    return 0 if result["matches"] else 1


def _cmd_snapshot_diff(args: argparse.Namespace) -> int:
    from aborist.snapshot import diff_against_current

    conn = (
        connect_query(args.db, shards_dir=args.global_shards_dir)
        if args.global_shards_dir
        else connect(args.db)
    )
    try:
        result = diff_against_current(conn, args.snapshot_root)
    finally:
        conn.close()
    print(json.dumps(result, indent=2, ensure_ascii=False))
    return 0


def _cmd_mesh_status(args: argparse.Namespace) -> int:
    from aborist.mesh import current_epoch, is_enabled, load_identity
    from aborist.mesh.state import roster_at

    conn = connect(args.db)
    try:
        ident = load_identity(conn)
        epoch = current_epoch(conn)
        roster = roster_at(conn, epoch) if epoch is not None else []
        out = {
            "enabled": is_enabled(conn),
            "identity": (
                {
                    "member_id": ident.member_id,
                    "group_name": ident.group_name,
                    "sign_pub_hex": ident.sign_pub.hex(),
                    "dh_pub_hex": ident.dh_pub.hex(),
                    "created_at": ident.created_at,
                }
                if ident
                else None
            ),
            "current_epoch": epoch,
            "roster": [
                {
                    "member_id": m.member_id,
                    "role": m.role,
                    "sign_pub_hex": m.sign_pub.hex(),
                    "dh_pub_hex": m.dh_pub.hex(),
                }
                for m in roster
            ],
        }
    finally:
        conn.close()
    print(json.dumps(out, indent=2, ensure_ascii=False))
    return 0


def _cmd_mesh_init(args: argparse.Namespace) -> int:
    from aborist.mesh import init_identity

    conn = connect(args.db)
    try:
        ident = init_identity(conn, group_name=args.group, member_id=args.member_id)
    except RuntimeError as e:
        print(f"error: {e}", file=sys.stderr)
        conn.close()
        return 2
    finally:
        conn.close()
    print(
        json.dumps(
            {
                "member_id": ident.member_id,
                "group_name": ident.group_name,
                "sign_pub_hex": ident.sign_pub.hex(),
                "dh_pub_hex": ident.dh_pub.hex(),
                "note": "share sign_pub_hex + dh_pub_hex with the founder of any group "
                "you want to join. Run 'mesh enable' to flip the gating flag on.",
            },
            indent=2, ensure_ascii=False
        )
    )
    return 0


def _cmd_mesh_enable(args: argparse.Namespace) -> int:
    from aborist.mesh import set_enabled

    conn = connect(args.db)
    try:
        set_enabled(conn, True)
    finally:
        conn.close()
    print(json.dumps({"enabled": True}, indent=2, ensure_ascii=False))
    return 0


def _cmd_mesh_disable(args: argparse.Namespace) -> int:
    from aborist.mesh import set_enabled

    conn = connect(args.db)
    try:
        set_enabled(conn, False)
    finally:
        conn.close()
    print(json.dumps({"enabled": False}, indent=2, ensure_ascii=False))
    return 0


def _cmd_mesh_members(args: argparse.Namespace) -> int:
    from aborist.mesh import current_epoch
    from aborist.mesh.state import roster_at

    conn = connect(args.db)
    try:
        epoch = current_epoch(conn)
        if epoch is None:
            print(json.dumps({"error": "mesh not initialized"}, indent=2, ensure_ascii=False))
            return 2
        roster = roster_at(conn, epoch)
    finally:
        conn.close()
    print(
        json.dumps(
            {
                "epoch": epoch,
                "members": [
                    {
                        "member_id": m.member_id,
                        "role": m.role,
                        "sign_pub_hex": m.sign_pub.hex(),
                        "dh_pub_hex": m.dh_pub.hex(),
                    }
                    for m in roster
                ],
            },
            indent=2, ensure_ascii=False
        )
    )
    return 0


def _cmd_mesh_add(args: argparse.Namespace) -> int:
    from aborist.mesh.members import add_member

    try:
        sign_pub = bytes.fromhex(args.sign_pub)
        dh_pub = bytes.fromhex(args.dh_pub)
    except ValueError:
        print("error: --sign-pub and --dh-pub must be hex-encoded 32-byte keys", file=sys.stderr)
        return 2
    if len(sign_pub) != 32 or len(dh_pub) != 32:
        print("error: keys must decode to exactly 32 bytes", file=sys.stderr)
        return 2

    conn = connect(args.db)
    try:
        epoch = add_member(
            conn,
            member_id=args.member_id,
            sign_pub=sign_pub,
            dh_pub=dh_pub,
            role=args.role,
        )
    except (PermissionError, RuntimeError, ValueError) as e:
        print(f"error: {e}", file=sys.stderr)
        return 2
    finally:
        conn.close()
    print(json.dumps({"new_epoch": epoch, "added": args.member_id}, indent=2, ensure_ascii=False))
    return 0


def _cmd_mesh_kick(args: argparse.Namespace) -> int:
    from aborist.mesh.members import kick_member

    conn = connect(args.db)
    try:
        epoch = kick_member(conn, member_id=args.member_id, reason=args.reason)
    except (PermissionError, RuntimeError, ValueError) as e:
        print(f"error: {e}", file=sys.stderr)
        return 2
    finally:
        conn.close()
    print(
        json.dumps(
            {"new_epoch": epoch, "kicked": args.member_id, "reason": args.reason},
            indent=2, ensure_ascii=False
        )
    )
    return 0


def _cmd_mesh_rotate(args: argparse.Namespace) -> int:
    from aborist.mesh.members import scheduled_rotate

    conn = connect(args.db)
    try:
        epoch = scheduled_rotate(conn, reason=args.reason)
    except (PermissionError, RuntimeError) as e:
        print(f"error: {e}", file=sys.stderr)
        return 2
    finally:
        conn.close()
    print(json.dumps({"new_epoch": epoch, "reason": args.reason}, indent=2, ensure_ascii=False))
    return 0


def _cmd_mesh_serve(args: argparse.Namespace) -> int:
    """Run the HTTP gossip server until SIGINT."""
    from aborist.mesh import is_enabled, load_identity
    from aborist.mesh.wire import MeshWireServer

    conn = connect(args.db)
    try:
        if load_identity(conn) is None:
            print("error: mesh not initialized; run 'aborist mesh init' first", file=sys.stderr)
            return 2
        if not is_enabled(conn):
            print("error: mesh.enabled is off; run 'aborist mesh enable' first", file=sys.stderr)
            return 2
    finally:
        conn.close()

    srv = MeshWireServer(args.db, host=args.host, port=args.port)
    print(json.dumps({"status": "serving", "url": srv.url, "db": str(args.db)}, ensure_ascii=False))
    sys.stdout.flush()
    try:
        srv.serve()
    except KeyboardInterrupt:
        print(json.dumps({"status": "stopped", "reason": "SIGINT"}, ensure_ascii=False))
    finally:
        srv.stop()
    return 0


def _cmd_mesh_sync(args: argparse.Namespace) -> int:
    """Push local roots + falsifications to a peer.

    Two pushes happen by default (unless ``--no-roots`` /
    ``--no-falsifications`` opts one out):

    1. **ANNOUNCE_ROOT** for the most-recent ``--limit`` documents
       older than the kindergarten window. Receivers dedup by
       ``documents.document_root``.
    2. **ANNOUNCE_FALSIFICATION** for the most-recent ``--limit``
       falsifications older than the kindergarten window. Burns
       deliberately NOT propagated — local kindergarten cleanup.

    **Kindergarten window.** Records younger than
    ``--kindergarten-seconds`` (default 3600 = 1 hour) are NOT
    broadcast. Gives the operator time to inspect a fresh ingest or
    falsification & burn it before the network sees it. Override per
    invocation; ``--kindergarten-seconds 0`` broadcasts everything
    (cron-friendly opt-out for operators who prefer immediate
    propagation). The window is sender-side discipline; receivers
    don't enforce it because they have no view into when the sender
    created the record.

    Receivers verify the Ed25519 signature, run per-peer chain-of-
    claims fork detection, then write one ``mesh_received`` audit
    event per accepted envelope. Duplicate broadcasts produce
    duplicate audit-log entries on the receiver but no state
    corruption.
    """
    import time as _time

    from aborist.mesh import is_enabled, load_identity
    from aborist.mesh.wire import MeshWireClient

    now_ts = int(_time.time())
    cutoff_ts = now_ts - max(0, args.kindergarten_seconds)

    conn = connect(args.db)
    try:
        if load_identity(conn) is None:
            print("error: mesh not initialized", file=sys.stderr)
            return 2
        if not is_enabled(conn):
            print("error: mesh.enabled is off", file=sys.stderr)
            return 2
        # Total counts inform skipped-by-kindergarten reporting.
        total_roots = 0
        total_falsifications = 0
        root_rows: list = []
        falsification_rows: list = []
        if not args.no_roots:
            total_roots = conn.execute(
                "SELECT COUNT(*) FROM documents"
            ).fetchone()[0]
            root_rows = conn.execute(
                "SELECT document_root, document_uri, chunking_version, "
                " canonicalization_version, schema_version "
                "FROM documents WHERE ingest_ts <= ? "
                "ORDER BY rowid DESC LIMIT ?",
                (cutoff_ts, args.limit),
            ).fetchall()
        if not args.no_falsifications:
            total_falsifications = conn.execute(
                "SELECT COUNT(*) FROM falsifications"
            ).fetchone()[0]
            falsification_rows = conn.execute(
                "SELECT cache_key, reason FROM falsifications "
                "WHERE at <= ? "
                "ORDER BY at DESC LIMIT ?",
                (cutoff_ts, args.limit),
            ).fetchall()
    finally:
        conn.close()

    # Skipped-by-kindergarten = (rows younger than cutoff that would have
    # been in the most-recent --limit) — approximated by total minus what
    # we pulled, capped at limit.
    fresh_roots_held = max(
        0,
        min(total_roots, args.limit) - len(root_rows),
    ) if not args.no_roots else 0
    fresh_falsifications_held = max(
        0,
        min(total_falsifications, args.limit) - len(falsification_rows),
    ) if not args.no_falsifications else 0

    sent_roots: list[dict] = []
    sent_falsifications: list[dict] = []
    errors: list[dict] = []
    with MeshWireClient(args.db, args.peer) as client:
        try:
            peer_info = client.info()
        except Exception as e:
            print(json.dumps({"status": "peer_unreachable", "peer": args.peer, "error": str(e)}, indent=2, ensure_ascii=False))
            return 2
        for r in root_rows:
            try:
                resp = client.announce_root(
                    document_root=r["document_root"],
                    source_uri=r["document_uri"],
                    chunking_version=r["chunking_version"],
                    canonicalization_version=r["canonicalization_version"],
                    schema_version=r["schema_version"],
                )
                sent_roots.append({"document_root": r["document_root"], "ack": resp})
            except Exception as e:
                errors.append({"document_root": r["document_root"], "error": str(e)})
        for f in falsification_rows:
            try:
                resp = client.announce_falsification(
                    cache_key=f["cache_key"],
                    reason=f["reason"] or "",
                )
                sent_falsifications.append({"cache_key": f["cache_key"], "ack": resp})
            except Exception as e:
                errors.append({"cache_key": f["cache_key"], "error": str(e)})

    print(json.dumps({
        "status": "synced",
        "peer": args.peer,
        "peer_member_id": peer_info.get("member_id"),
        "peer_epoch": peer_info.get("current_epoch"),
        "kindergarten_seconds": args.kindergarten_seconds,
        "announced_roots": len(sent_roots),
        "announced_falsifications": len(sent_falsifications),
        "kindergarten_held_roots": fresh_roots_held,
        "kindergarten_held_falsifications": fresh_falsifications_held,
        "errors": len(errors),
        "sent_roots": sent_roots[: args.verbose],
        "sent_falsifications": sent_falsifications[: args.verbose],
        "error_samples": errors[:5],
    }, indent=2, ensure_ascii=False))
    return 0 if not errors else 1


def _cmd_mesh_pull(args: argparse.Namespace) -> int:
    """Pull a single document body from a peer by document_root.

    Closes the request half of the gossip loop. The wire client already
    verifies the peer's signature and re-derives the Merkle root from the
    delivered leaves before returning. This verb then re-ingests the
    delivered text through the standard ingest path so chunking_version /
    canonicalization_version stay consistent — and rejects with rc=2 if
    the local re-ingest produces a different document_root than requested.
    """
    from aborist.document import Document
    from aborist.ingest import ingest_source
    from aborist.mesh import is_enabled, load_identity
    from aborist.mesh.wire import MeshWireClient

    conn = connect(args.db)
    try:
        if load_identity(conn) is None:
            print("error: mesh not initialized", file=sys.stderr)
            return 2
        if not is_enabled(conn):
            print("error: mesh.enabled is off", file=sys.stderr)
            return 2
        already = conn.execute(
            "SELECT document_root, document_uri FROM documents WHERE document_root=?",
            (args.root,),
        ).fetchone()
    finally:
        conn.close()

    if already is not None:
        print(json.dumps({
            "status": "already_present",
            "document_root": already["document_root"],
            "document_uri": already["document_uri"],
            "shard": str(args.db),
        }, indent=2, ensure_ascii=False))
        return 0

    try:
        with MeshWireClient(args.db, args.peer) as client:
            body = client.request_body(root=args.root)
    except Exception as e:
        print(f"error: pull failed: {e}", file=sys.stderr)
        return 2

    delivered_uri = body.get("document_uri") or ""
    delivered_text = body.get("text") or ""

    class _PulledSource:
        source_type = "mesh_pull"

        def iter_documents(self):
            yield Document(
                uri=delivered_uri,
                content=delivered_text,
                source_type="mesh_pull",
                title=None,
            )

    conn = connect(args.db)
    try:
        ingest_source(conn, _PulledSource())
        row = conn.execute(
            "SELECT document_root FROM documents WHERE document_root=?",
            (args.root,),
        ).fetchone()
        if row is None:
            # Re-ingest produced a different root than the peer claimed.
            # The pulled text doesn't reproduce the requested root under
            # this peer's chunker/canonicalization. Fail closed.
            actual = conn.execute(
                "SELECT document_root FROM documents WHERE document_uri=? "
                "ORDER BY ingest_ts DESC LIMIT 1",
                (delivered_uri,),
            ).fetchone()
            actual_root = actual["document_root"] if actual else None
            print(
                "error: local re-ingest produced "
                f"{actual_root!r}, expected {args.root!r}",
                file=sys.stderr,
            )
            return 2
        with transaction(conn):
            event_hash = append_audit(
                conn,
                event_type="mesh_pulled",
                body={
                    "document_root": args.root,
                    "document_uri": delivered_uri,
                    "peer": args.peer,
                },
                subject_root=args.root,
            )
    finally:
        conn.close()

    print(json.dumps({
        "status": "pulled",
        "document_root": args.root,
        "document_uri": delivered_uri,
        "shard": str(args.db),
        "audit_event_hash": event_hash,
    }, indent=2, ensure_ascii=False))
    return 0


def _cmd_crawl(args: argparse.Namespace) -> int:
    """BFS-discover same-domain URLs from a seed and optionally ingest.

    Two modes:

    - default: print discovered URLs to stdout (one per line). Compose
      with `aborist ingest --source html` if you want to feed them
      through the standard ingest path manually.
    - ``--ingest``: run the discovery + ingest path in a single shot,
      capturing ETag + Last-Modified per page so a future
      ``crawler recrawl-check`` can do conditional HEADs.
    """
    try:
        from aborist.sources.crawler.bridge import crawl_seed, ingest_crawled
    except ImportError as e:
        print(f"error: {e}", file=sys.stderr)
        return 2

    from aborist.progress import Progress

    cap = "no cap" if args.max_pages == 0 else f"max {args.max_pages}"
    speed = "fast" if args.fast else "polite"
    print(
        f"  crawl: seed={args.seed_url} depth={args.depth} {cap} ({speed})",
        file=sys.stderr,
        flush=True,
    )
    crawl_progress = Progress(prefix="crawl ")
    urls = crawl_seed(
        args.seed_url,
        max_depth=args.depth,
        max_pages=args.max_pages,
        progress=crawl_progress,
        fast=args.fast,
    )
    print(
        f"  crawl: discovery done — {len(urls)} URLs",
        file=sys.stderr,
        flush=True,
    )

    if not args.ingest:
        for u in urls:
            print(u)
        return 0

    print(
        f"  ingest: starting on {len(urls)} URLs",
        file=sys.stderr,
        flush=True,
    )
    ingest_progress = Progress(prefix="ingest ", total_estimate=len(urls))
    conn = connect(args.db)
    try:
        result = ingest_crawled(conn, urls, progress=ingest_progress)
    finally:
        conn.close()
    print(json.dumps(
        {
            "status": "crawled_and_ingested",
            "seed": args.seed_url,
            "depth": args.depth,
            "max_pages": args.max_pages,
            "discovered": len(urls),
            **result,
        },
        indent=2, ensure_ascii=False
    ))
    return 0


def _cmd_crawler_recrawl_check(args: argparse.Namespace) -> int:
    """Send conditional HEAD requests for ingested documents.

    Reports each as fresh (304), stale (200, body changed), gone
    (404/410), or unreachable. Updates `document_http_meta.last_status`
    and `last_checked_at` so consecutive runs target the oldest checks
    first.
    """
    try:
        from aborist.sources.crawler.bridge import recrawl_check
    except ImportError as e:
        print(f"error: {e}", file=sys.stderr)
        return 2

    conn = connect(args.db)
    try:
        result = recrawl_check(
            conn,
            domain=args.domain,
            limit=args.limit,
        )
    finally:
        conn.close()
    print(json.dumps(result, indent=2, ensure_ascii=False))
    return 0


def build_parser() -> argparse.ArgumentParser:
    p = argparse.ArgumentParser(
        prog="aborist",
        description="An arborist for trees and forests of cross-linked information.",
    )
    p.add_argument("--version", action="version", version=f"aborist {__version__}")
    p.add_argument(
        "--db",
        type=Path,
        default=DEFAULT_DB_PATH,
        help=f"path to aborist SQLite db (default: {DEFAULT_DB_PATH})",
    )
    p.add_argument(
        "--shards-dir",
        dest="global_shards_dir",
        default=None,
        help=(
            "for read commands: attach all shards in this directory and "
            "expose them as UNION views over the standard tables"
        ),
    )
    sub = p.add_subparsers(dest="cmd", required=True)

    ingest = sub.add_parser("ingest", help="ingest documents from a source")
    ingest.add_argument(
        "--source",
        required=True,
        choices=[
            "wikipedia_cur",
            "wikipedia_old",
            "wikipedia_xml",
            "wikipedia_xml_history",
            "wikipedia_abstract",
            "html",
            "grok_export",
            "grok_media",
            "git_repo",
            "hg_repo",
            "providence",
        ],
        help="source type",
    )
    ingest.add_argument(
        "--kindergarten-seconds",
        type=int,
        default=None,
        help=(
            "(providence source only) records younger than this many "
            "seconds stay opaque to ingestion — fresh thoughts cool "
            "before they become substrate. Default 3600s (1h)."
        ),
    )
    ingest.add_argument(
        "--path",
        help=(
            "path to dump file (wikipedia) or to xAI export root / "
            "prod-grok-backend.json (grok_export, grok_media)"
        ),
    )
    ingest.add_argument(
        "--url", action="append", help="URL to ingest (html source; repeatable)"
    )
    ingest.add_argument(
        "--urls-from",
        dest="urls_from",
        help="file with one URL per line (html source)",
    )
    ingest.add_argument(
        "--no-robots",
        dest="no_robots",
        action="store_true",
        help="do not consult robots.txt (use only for explicitly authorized sites)",
    )
    ingest.add_argument(
        "--chunker", default=None, help="chunker name (default: tok-512-v1)"
    )
    ingest.add_argument(
        "--limit", type=int, default=None, help="cap number of documents"
    )
    ingest.add_argument(
        "--batch-size",
        dest="batch_size",
        type=int,
        default=200,
        help="documents per SQLite transaction (default 200)",
    )
    ingest.add_argument(
        "--shard",
        default=None,
        help=(
            "rank/total — yield only every N-th doc for parallel ingest. "
            "spawn N processes, each with --shard 0/N, 1/N, ... they "
            "parallelize parser CPU and serialize writes via WAL"
        ),
    )
    ingest.add_argument(
        "--shards-dir",
        dest="shards_dir",
        default=None,
        help=(
            "directory for attach-forever sharding. With --shard rank/total, "
            "writes to shards-dir/<rank>.db instead of --db, removing the "
            "WAL writer-lock contention entirely. Reads via aborist --shards-dir"
        ),
    )
    ingest.add_argument(
        "--resume",
        action="store_true",
        help=(
            "rsync-style: read each source's last high-water mark from this "
            "DB's meta table and skip rows whose id is <= it. Safe to kill "
            "and restart at any time"
        ),
    )
    ingest.add_argument(
        "--quiet",
        action="store_true",
        help="suppress periodic stderr progress output",
    )
    ingest.add_argument(
        "--progress-interval",
        dest="progress_interval",
        type=float,
        default=2.0,
        help="seconds between stderr progress lines (default 2.0)",
    )
    ingest.add_argument(
        "--total-estimate",
        dest="total_estimate",
        type=int,
        default=None,
        help=(
            "estimated total docs the source will yield. enables percent "
            "+ ETA in progress output"
        ),
    )
    ingest.set_defaults(func=_cmd_ingest)

    search = sub.add_parser("search", help="keyword search (UNGROUNDED audit mode)")
    search.add_argument("query", help="query string")
    search.add_argument("--limit", type=int, default=20)
    search.add_argument("--json", action="store_true", help="output JSON")
    search.set_defaults(func=_cmd_search)

    verify = sub.add_parser(
        "verify", help="round-trip Merkle proofs for N random documents"
    )
    verify.add_argument("-n", type=int, default=10)
    verify.set_defaults(func=_cmd_verify)

    distill = sub.add_parser(
        "distill",
        help="compress docs into Merkle-signed cores (surface->core or core->core)",
    )
    distill.add_argument(
        "--process", default="first-sentence-v1", help="distiller name"
    )
    distill.add_argument(
        "--kind",
        choices=["surface", "core"],
        default="surface",
        help="source kind to scan; 'core' runs recursive distillation",
    )
    distill.add_argument(
        "--source-type",
        dest="source_type",
        default=None,
        help="restrict to one source_type",
    )
    distill.add_argument(
        "--chunker", default=None, help="chunker for the core doc"
    )
    distill.add_argument(
        "--limit", type=int, default=None, help="cap number of docs scanned"
    )
    distill.add_argument(
        "--batch-size",
        dest="batch_size",
        type=int,
        default=200,
        help="cores written per SQLite transaction (default 200)",
    )
    distill.set_defaults(func=_cmd_distill)

    ask_cmd = sub.add_parser(
        "ask",
        help="answer a question about a document (cache-first, STRICT)",
    )
    ask_cmd.add_argument(
        "--document-root",
        dest="document_root",
        required=True,
        help="document_root to ask about",
    )
    ask_cmd.add_argument(
        "--question", required=True, help="question text"
    )
    ask_cmd.add_argument(
        "--model",
        default=None,
        help="model_id (default $ABORIST_LLM_MODEL or hermes-3)",
    )
    ask_cmd.add_argument(
        "--endpoint",
        default=None,
        help="OpenAI-compatible base URL (default $ABORIST_LLM_ENDPOINT)",
    )
    ask_cmd.add_argument(
        "--dry-run",
        dest="dry_run",
        action="store_true",
        help="use StubClient — no network call",
    )
    ask_cmd.add_argument(
        "--answer-mode", dest="answer_mode", default=None,
        choices=["quote", "claim_lattice_pointer", "claim_lattice"],
        help=(
            "answer schema. See `query --answer-mode` for full semantics. "
            "Default 'quote'; 'claim_lattice_pointer' enables quote-by-pointer; "
            "'claim_lattice' is the JSON variant (vLLM guided_json + lenient "
            "pre-parser; pairs with grammar-constrained inference)."
        ),
    )
    ask_cmd.set_defaults(func=_cmd_ask)

    query_cmd = sub.add_parser(
        "query",
        help="multi-source RAG: question -> top-K corpus docs -> Hermes -> cache",
    )
    query_cmd.add_argument("question", help="the question to ask")
    query_cmd.add_argument(
        "--top-k", dest="top_k", type=int, default=8,
        help="max distinct source documents in context (default 8)",
    )
    query_cmd.add_argument(
        "--over-fetch", dest="over_fetch", type=int, default=32,
        help="FTS5 hits to fetch per shard before dedup (default 32)",
    )
    query_cmd.add_argument(
        "--max-context-chars", dest="max_context_chars", type=int, default=None,
        help=(
            "cap on assembled context bytes. When omitted, falls back to "
            "the per-mode default in DEFAULT_QUERY_POLICY['max_context_chars_by_mode'] "
            "(quote=24000, claim_lattice_pointer=24000, claim_lattice=48000). "
            "Sprint 1b 2026-05-02 — peaks measured per mode."
        ),
    )
    query_cmd.add_argument(
        "--question-dedup", dest="question_dedup", default=None,
        choices=["strict", "equivalence_class"],
        help=(
            "write-time question canonicalization. 'equivalence_class' "
            "(default) collapses articles + trailing-punct + case so "
            "variants share cache_keys. 'strict' keeps every variant "
            "distinct (audit-grade)."
        ),
    )
    query_cmd.add_argument(
        "--fidelity", dest="fidelity", default=None,
        choices=["strict", "equivalence_class"],
        help=(
            "lookup tolerance. 'equivalence_class' (default) tries the "
            "primary cache_key then falls back to the alternate dedup "
            "mode's cache_key. 'strict' refuses fallback."
        ),
    )
    query_cmd.add_argument(
        "--qa-db", dest="qa_db", default=None,
        help=(
            "providence_cache target DB. default: <shards-dir>/qa.db, or "
            "~/.aborist/qa.db when no shards-dir"
        ),
    )
    query_cmd.add_argument(
        "--model", default=None,
        help="model_id (default $ABORIST_LLM_MODEL or hermes-3)",
    )
    query_cmd.add_argument(
        "--endpoint", default=None,
        help="OpenAI-compatible base URL (default $ABORIST_LLM_ENDPOINT)",
    )
    query_cmd.add_argument(
        "--dry-run", dest="dry_run", action="store_true",
        help="use StubClient — assembles context but skips the LLM call",
    )
    query_cmd.add_argument(
        "--json", action="store_true",
        help="emit the raw record as indented JSON (default: human render)",
    )
    query_cmd.add_argument(
        "--burn", action="store_true",
        help=(
            "delete the matching live providence_cache row BEFORE lookup, "
            "forcing a fresh inference. Test-ergonomic — see new behavior "
            "without finding cache_keys by hand. Writes a providence_burn "
            "audit event."
        ),
    )
    query_cmd.add_argument(
        "--repair", action="store_true",
        help=(
            "enable mechanical repair after first verify (off by "
            "default). When the verdict is HYBRID/UNGROUNDED, applies "
            "synthetic_elision split, trailing_artifact trim, and "
            "no_overlap drop deterministically; persists the repaired "
            "answer with a providence_repair audit event."
        ),
    )
    query_cmd.add_argument(
        "--repair-reprompts", dest="repair_reprompts", type=int, default=0,
        help=(
            "max LLM re-prompt iterations after mechanical repair "
            "(default 0 = no re-prompt). Each iteration sends a feedback "
            "turn naming failed quotes; the model is asked to rewrite "
            "using only verbatim citations. Requires --repair."
        ),
    )
    query_cmd.add_argument(
        "--answer-mode", dest="answer_mode", default=None,
        choices=["quote", "claim_lattice_pointer", "claim_lattice"],
        help=(
            "answer schema. 'quote' (default): model writes prose with "
            "verbatim quote spans inline. 'claim_lattice_pointer' (G0 "
            "/ CTI quote-by-pointer): runtime builds a labeled evidence "
            "map (E1, E2, …); model writes pointer-line prose ('Claim. "
            "[E12]'); renderer interpolates literal spans. Synthetic-"
            "elision-by-construction-impossible. No repair loop "
            "(one-shot discipline). 'claim_lattice' is the JSON variant "
            "(vLLM guided_json + lenient pre-parser; pairs with grammar-"
            "constrained inference like Qwen 3.6 reasoner / Claude / GPT-4)."
        ),
    )
    query_cmd.add_argument(
        "--retrieval-keywords", dest="retrieval_keywords", default=None,
        help=(
            "operator-supplied keywords appended to the question for "
            "FTS5 retrieval ONLY — never sent to the LLM, never enters "
            "cache_key, never reaches the verifier. Use to narrow "
            "OR-mode retrieval on long discursive questions whose "
            "content tokens get diluted by template phrasing. Example: "
            "make query Q='what tech may enable one person to "
            "reconstruct another person's thoughts...' "
            "K='transcranial knowledge acquisition'. Pair with --burn "
            "to force fresh inference (keywords are session-only and "
            "cache-hits ignore them)."
        ),
    )
    # Ticket #000008 Phase 4 — quantifier-guard CLI flags. Each
    # corresponds to a level of the §10.11.2 disable hierarchy.
    query_cmd.add_argument(
        "--no-quantifier-guard",
        dest="no_quantifier_guard", action="store_true",
        help=(
            "Disable the broad-quantifier preflight guard for this "
            "call. Overrides quantifier_guard_enabled in policy. "
            "Bench-side telemetry (quantifier_intensity, etc.) goes "
            "to None for the row. Use when the guard misclassifies."
        ),
    )
    query_cmd.add_argument(
        "--allow-broad",
        dest="allow_broad", action="store_true",
        help=(
            "Emergent-search mode: keep the classifier on (telemetry "
            "stays useful) but don't apply caps. For broad questions "
            "where the operator wants exploratory enumeration, not "
            "grounded completeness."
        ),
    )
    query_cmd.add_argument(
        "--reject-broad",
        dest="reject_broad", action="store_true",
        help=(
            "Strict mode: when intensity is ALL/COMPREHENSIVE/"
            "OPEN_REQUEST AND scope_bound_hint is unbounded, return "
            "UNGROUNDED before the LLM call with a "
            "BROAD_QUANTIFIER_REJECTED violation. Saves ~10-15s on "
            "rejected runs. Bounded universals (e.g. all members of "
            "the Beatles) are NOT rejected."
        ),
    )
    query_cmd.add_argument(
        "--apply-quantifier-caps",
        dest="apply_quantifier_caps", action="store_true",
        help=(
            "Flip the dry-run gate per-call. By default Phase 2 "
            "lands with quantifier_guard_apply_caps=False so the "
            "cap is reported on the result but not applied to the "
            "verifier. This flag enables actual cap enforcement "
            "for one call. Use after dry-run bench review confirms "
            "the classifier output across the question set."
        ),
    )
    # Ticket #000010 — meta-cognition CLI flags.
    query_cmd.add_argument(
        "--no-preflight",
        dest="no_preflight", action="store_true",
        help=(
            "Disable the meta-cognition preflight guard for this "
            "call. Skips temporal / contradiction / false-premise / "
            "out-of-corpus detectors. The QuestionState surfaces a "
            "stub with empty logical_statuses so bench rows stay "
            "column-aligned."
        ),
    )
    query_cmd.add_argument(
        "--block-on-contradiction",
        dest="block_on_contradiction", action="store_true",
        help=(
            "Hard-block on lexical contradictions (default: label-"
            "only). Strict mode: questions like 'which unmarried "
            "spouse is X married to' return PREFLIGHT_BLOCKED."
        ),
    )
    query_cmd.add_argument(
        "--soft-preflight",
        dest="soft_preflight", action="store_true",
        help=(
            "Ticket #000011 — opt-in to the model-assisted soft "
            "preflight sidecar. Adds one short LLM round-trip "
            "(~200ms median) before the main answer call; the model "
            "classifies the question shape and returns a SOFT_* "
            "advisory hint that surfaces as `· soft: <label>` on "
            "the audit-line tail. NEVER enters the verifier proof "
            "path; cannot create PREFLIGHT_OK or PREFLIGHT_BLOCKED."
        ),
    )
    query_cmd.set_defaults(func=_cmd_query)

    inspect_cmd = sub.add_parser(
        "inspect",
        help=(
            "sidecar diagnostic for a providence_cache record — pulls "
            "source chunks and classifies each unverified span "
            "(paraphrase / trailing_artifact / interior_elision / "
            "synthetic_elision_inside_quote / "
            "no_overlap). Read-only, "
            "no audit events, no v9.8 field changes."
        ),
    )
    inspect_cmd.add_argument(
        "--cache-key", dest="cache_key", required=True,
        help="64-char hex cache_key of the providence record to inspect",
    )
    inspect_cmd.add_argument(
        "--qa-db", dest="qa_db", default=None,
        help="path to qa.db (default: <shards>/qa.db or ~/.aborist/qa.db)",
    )
    inspect_cmd.add_argument(
        "--json", action="store_true",
        help="emit raw diagnosis as JSON (default: human render)",
    )
    inspect_cmd.set_defaults(func=_cmd_inspect)

    prov_cmd = sub.add_parser(
        "providence",
        help="list or falsify providence_cache records",
    )
    prov_cmd.add_argument("--document-uri", dest="document_uri", default=None)
    prov_cmd.add_argument("--source-root", dest="source_root", default=None)
    prov_cmd.add_argument("--limit", type=int, default=20)
    prov_cmd.add_argument(
        "--falsify",
        default=None,
        metavar="CACHE_KEY",
        help=(
            "mark a providence_cache record as failed/stale/quarantined. "
            "Lookups will skip it. Audit chain records the act"
        ),
    )
    prov_cmd.add_argument(
        "--state",
        default="failed",
        choices=["failed", "stale", "quarantined"],
        help="falsification state to set (default: failed)",
    )
    prov_cmd.add_argument(
        "--reason",
        default=None,
        help="reason text stored in falsifications log",
    )
    prov_cmd.add_argument(
        "--by-actor",
        dest="by_actor",
        default=None,
        help="who is falsifying (default: $USER)",
    )
    prov_cmd.add_argument(
        "--show-preflight",
        dest="show_preflight",
        default=None,
        metavar="CACHE_KEY_PREFIX",
        help=(
            "Pull the preflight stage payload from a row's "
            "run_dag_blob. Match by 12-char prefix. Renders the "
            "preflight stage hash + run-DAG stage list. Operator "
            "tool for inspecting the policy state that governed a "
            "cached row (#000009 §7.2)."
        ),
    )
    prov_cmd.set_defaults(func=_cmd_providence)

    burn_cmd = sub.add_parser(
        "burn",
        help=(
            "delete a leaf with no children — providence_cache, document, "
            "or core (kindergarten use; falsify/evict are audit-preserving)"
        ),
    )
    burn_cmd.add_argument(
        "--kind",
        choices=("providence", "document", "core"),
        default="providence",
        help="leaf kind to burn (default: providence — backwards-compatible)",
    )
    burn_cmd.add_argument(
        "--cache-key",
        dest="cache_key",
        default=None,
        help="cache_key (hex) of the providence record to burn (kind=providence)",
    )
    burn_cmd.add_argument(
        "--root",
        dest="root",
        default=None,
        help="document_root (hex) of the document/core to burn (kind=document|core)",
    )
    burn_cmd.add_argument(
        "--reason",
        default=None,
        help="reason text recorded in the burn audit event",
    )
    burn_cmd.add_argument(
        "--by-actor",
        dest="by_actor",
        default=None,
        help="who is burning (default: $USER)",
    )
    burn_cmd.add_argument(
        "--force",
        action="store_true",
        help="burn even if children exist; not recommended",
    )
    burn_cmd.set_defaults(func=_cmd_burn)

    burn_kg_cmd = sub.add_parser(
        "burn-kindergarten",
        help=(
            "burn every providence_cache record younger than the "
            "kindergarten window — test-ergonomic mass cleanup that "
            "matches the mesh-sync kindergarten window so only "
            "un-broadcast records get busted"
        ),
    )
    burn_kg_cmd.add_argument(
        "--kindergarten-seconds",
        dest="kindergarten_seconds",
        type=int,
        default=3600,
        help=(
            "burn rows younger than this many seconds (default: 3600 = 1 "
            "hour, mirrors mesh sync default). 0 = burn everything live."
        ),
    )
    burn_kg_cmd.add_argument(
        "--reason", default=None,
        help="reason text recorded in each providence_burn audit event",
    )
    burn_kg_cmd.add_argument(
        "--by-actor", dest="by_actor", default=None,
        help="who is burning (default: $USER)",
    )
    burn_kg_cmd.add_argument(
        "--force", action="store_true",
        help="burn even if rows have falsification children",
    )
    burn_kg_cmd.add_argument(
        "--dry-run", dest="dry_run", action="store_true",
        help="report what would burn without writing",
    )
    burn_kg_cmd.add_argument(
        "--verbose", type=int, default=10,
        help="include this many items in the result JSON (default: 10)",
    )
    burn_kg_cmd.set_defaults(func=_cmd_burn_kindergarten)

    reclassify_cmd = sub.add_parser(
        "reclassify",
        help="re-run the verifier against existing live providence records "
             "(no LLM calls; relabels stale classifications)",
    )
    reclassify_cmd.add_argument(
        "--qa-db", dest="qa_db", default=None,
        help="path to qa.db (default: <shards>/qa.db or ~/.aborist/qa.db)",
    )
    reclassify_cmd.add_argument(
        "--limit", type=int, default=0,
        help="reclassify at most N records (0 = unlimited)",
    )
    reclassify_cmd.add_argument(
        "--dry-run", dest="dry_run", action="store_true",
        help="report what would change without writing",
    )
    reclassify_cmd.add_argument(
        "--entity-policy", dest="entity_policy", default=None,
        choices=["strict", "hybrid", "drop", "proximity"],
        help=(
            "how the entity path classifies: 'strict' (legacy, overclaims), "
            "'hybrid' (default — caps at HYBRID), 'drop' (skip entity path → "
            "UNGROUNDED), 'proximity' (STRICT only if N entities cluster within "
            "W chars in source)"
        ),
    )
    reclassify_cmd.add_argument(
        "--compare", dest="compare", action="store_true",
        help="run all four entity policies side-by-side without writing",
    )
    reclassify_cmd.set_defaults(func=_cmd_reclassify)

    emergent_cmd = sub.add_parser(
        "emergent",
        help="surface UNGROUNDED/HYBRID claims — corpus-growth signal",
    )
    emergent_cmd.add_argument(
        "--aggregate",
        action="store_true",
        help="rank unverified quotes by frequency (vs per-record list)",
    )
    emergent_cmd.add_argument("--limit", type=int, default=20)
    emergent_cmd.set_defaults(func=_cmd_emergent)

    evict_cmd = sub.add_parser(
        "evict",
        help="demote surface chunks hot→cold (NULL content, retain leaf_hash)",
    )
    evict_cmd.add_argument(
        "--source-type",
        dest="source_type",
        default=None,
        help="restrict to one source_type",
    )
    evict_cmd.add_argument(
        "--older-than-days",
        dest="older_than_days",
        type=int,
        default=None,
        help="only evict docs older than N days",
    )
    evict_cmd.add_argument(
        "--document-root",
        action="append",
        default=None,
        help="explicit document_root(s) to evict; repeatable",
    )
    evict_cmd.set_defaults(func=_cmd_evict)

    rehydrate_cmd = sub.add_parser(
        "rehydrate",
        help="refetch URI, verify leaves, restore cold content if root matches",
    )
    rehydrate_cmd.add_argument(
        "--document-root",
        action="append",
        default=None,
        help="explicit document_root(s) to rehydrate; repeatable",
    )
    rehydrate_cmd.add_argument(
        "--all-cold",
        dest="all_cold",
        action="store_true",
        help="rehydrate every document with cold chunks",
    )
    rehydrate_cmd.set_defaults(func=_cmd_rehydrate)

    activity_cmd = sub.add_parser(
        "activity",
        help="recent Q&A + freshly cached docs (agent-readable timeline)",
    )
    activity_cmd.add_argument(
        "--limit", type=int, default=10,
        help="max items per category (default 10)",
    )
    activity_cmd.add_argument(
        "--since-seconds",
        dest="since_seconds",
        type=int,
        default=0,
        help="only events newer than this many seconds (0 = all time, default)",
    )
    activity_cmd.add_argument(
        "--preview-chars",
        dest="preview_chars",
        type=int,
        default=240,
        help="answer preview length (default 240 chars)",
    )
    activity_cmd.set_defaults(func=_cmd_activity)

    stats_cmd = sub.add_parser("stats", help="counts: docs, chunks, edges, audit")
    stats_cmd.set_defaults(func=_cmd_stats)

    analyze_cmd = sub.add_parser(
        "analyze",
        help="compression spectrum, depth distribution, audit chain integrity",
    )
    analyze_cmd.add_argument(
        "--gravity-top",
        dest="gravity_top",
        type=int,
        default=10,
        help="N top inbound-linked URIs to report (default 10)",
    )
    analyze_cmd.set_defaults(func=_cmd_analyze)

    # ----- snapshot subcommands ----------------------------------------------
    snap_cmd = sub.add_parser(
        "snapshot",
        help="corpus-level Merkle snapshots: pin a forest state by single root",
    )
    snap_sub = snap_cmd.add_subparsers(dest="snap_op", required=True)

    snap_create = snap_sub.add_parser(
        "create", help="compute snapshot_root from current corpus, persist + audit"
    )
    snap_create.add_argument("--reason", default="manual")
    snap_create.add_argument(
        "--parent",
        default=None,
        help="explicit parent_snapshot hex (default: auto-link to latest prior snapshot)",
    )
    snap_create.set_defaults(func=_cmd_snapshot_create)

    snap_list = snap_sub.add_parser("list", help="recent snapshots, newest first")
    snap_list.add_argument("--limit", type=int, default=20)
    snap_list.set_defaults(func=_cmd_snapshot_list)

    snap_verify = snap_sub.add_parser(
        "verify",
        help="recompute root from current corpus; matches=True iff nothing has changed",
    )
    snap_verify.add_argument("snapshot_root", help="hex snapshot_root to verify")
    snap_verify.set_defaults(func=_cmd_snapshot_verify)

    snap_diff = snap_sub.add_parser(
        "diff",
        help="coarse drift signal between a snapshot and the current corpus",
    )
    snap_diff.add_argument("snapshot_root", help="hex snapshot_root to diff against current")
    snap_diff.set_defaults(func=_cmd_snapshot_diff)

    # ----- mesh subcommands (off by default) ---------------------------------
    mesh_cmd = sub.add_parser(
        "mesh",
        help="federation/gossip layer (off by default; opt-in via 'mesh enable')",
    )
    mesh_sub = mesh_cmd.add_subparsers(dest="mesh_op", required=True)

    mesh_status = mesh_sub.add_parser("status", help="show enabled flag, identity, current epoch + roster")
    mesh_status.set_defaults(func=_cmd_mesh_status)

    mesh_init = mesh_sub.add_parser("init", help="generate this peer's keys; create epoch 0")
    mesh_init.add_argument("--group", required=True, help="group name")
    mesh_init.add_argument("--member-id", dest="member_id", default=None, help="optional fixed member id (default: random 8-hex)")
    mesh_init.set_defaults(func=_cmd_mesh_init)

    mesh_enable = mesh_sub.add_parser("enable", help="flip the mesh.enabled flag on")
    mesh_enable.set_defaults(func=_cmd_mesh_enable)

    mesh_disable = mesh_sub.add_parser("disable", help="flip the mesh.enabled flag off")
    mesh_disable.set_defaults(func=_cmd_mesh_disable)

    mesh_members = mesh_sub.add_parser("members", help="list current epoch's roster")
    mesh_members.set_defaults(func=_cmd_mesh_members)

    mesh_add = mesh_sub.add_parser("add", help="admin-only: add a peer to the roster (bumps epoch)")
    mesh_add.add_argument("--member-id", dest="member_id", required=True)
    mesh_add.add_argument("--sign-pub", dest="sign_pub", required=True, help="hex Ed25519 pubkey (32 bytes / 64 hex chars)")
    mesh_add.add_argument("--dh-pub", dest="dh_pub", required=True, help="hex X25519 pubkey")
    mesh_add.add_argument("--role", choices=["admin", "member"], default="member")
    mesh_add.set_defaults(func=_cmd_mesh_add)

    mesh_kick = mesh_sub.add_parser("kick", help="admin-only: evict a peer (bumps epoch; old signatures stay valid, new gossip is opaque to them)")
    mesh_kick.add_argument("--member-id", dest="member_id", required=True)
    mesh_kick.add_argument("--reason", required=True)
    mesh_kick.set_defaults(func=_cmd_mesh_kick)

    mesh_rotate = mesh_sub.add_parser("rotate", help="refresh epoch secret without changing roster")
    mesh_rotate.add_argument("--reason", default="scheduled")
    mesh_rotate.set_defaults(func=_cmd_mesh_rotate)

    mesh_serve = mesh_sub.add_parser(
        "serve",
        help="run the HTTP gossip server (blocks until SIGINT)",
    )
    mesh_serve.add_argument("--host", default="127.0.0.1", help="bind host (default: 127.0.0.1)")
    mesh_serve.add_argument("--port", type=int, default=8400, help="bind port (default: 8400)")
    mesh_serve.set_defaults(func=_cmd_mesh_serve)

    mesh_sync = mesh_sub.add_parser(
        "sync",
        help="announce local document_roots to a peer's gossip server",
    )
    mesh_sync.add_argument("--peer", required=True, help="peer URL, e.g. http://other.example.com:8400")
    mesh_sync.add_argument("--limit", type=int, default=100, help="announce at most N most-recent items per category (default: 100)")
    mesh_sync.add_argument("--verbose", type=int, default=10, help="include this many ack details in output (default: 10)")
    mesh_sync.add_argument(
        "--no-roots",
        dest="no_roots",
        action="store_true",
        help="skip ANNOUNCE_ROOT broadcast (only push falsifications)",
    )
    mesh_sync.add_argument(
        "--no-falsifications",
        dest="no_falsifications",
        action="store_true",
        help="skip ANNOUNCE_FALSIFICATION broadcast (only push roots)",
    )
    mesh_sync.add_argument(
        "--kindergarten-seconds",
        dest="kindergarten_seconds",
        type=int,
        default=3600,
        help=(
            "hold records younger than this many seconds back from the "
            "broadcast (default: 3600 = 1 hour). Gives operators time to "
            "burn or falsify before peers see it. 0 = broadcast everything."
        ),
    )
    mesh_sync.set_defaults(func=_cmd_mesh_sync)

    mesh_pull = mesh_sub.add_parser(
        "pull",
        help="pull one document body from a peer by document_root",
    )
    mesh_pull.add_argument("--root", required=True, help="64-char hex document_root to pull")
    mesh_pull.add_argument("--peer", required=True, help="peer URL, e.g. http://other.example.com:8400")
    mesh_pull.set_defaults(func=_cmd_mesh_pull)

    crawl_cmd = sub.add_parser(
        "crawl",
        help=(
            "BFS-discover same-domain URLs from a seed; optionally ingest "
            "and store ETag/Last-Modified for cheap recrawl-checks "
            "(requires aborist[crawler] extras)"
        ),
    )
    crawl_cmd.add_argument("--seed-url", dest="seed_url", required=True)
    crawl_cmd.add_argument("--depth", type=int, default=2, help="max BFS depth (default: 2)")
    crawl_cmd.add_argument(
        "--max-pages",
        dest="max_pages",
        type=int,
        default=0,
        help="cap discovery at N URLs (0 = no cap, depth is the only bound; default: 0)",
    )
    crawl_cmd.add_argument(
        "--ingest",
        action="store_true",
        help="ingest the discovered pages into --db (default: print URL list only)",
    )
    crawl_cmd.add_argument(
        "--fast",
        action="store_true",
        help=(
            "fast_mode: 5s timeouts, CPU*3 parallel page workers, ignore "
            "robots.txt crawl-delay (Disallow is still honored). Use only "
            "against domains where aggressive fetching is acceptable."
        ),
    )
    crawl_cmd.set_defaults(func=_cmd_crawl)

    crawler_cmd = sub.add_parser(
        "crawler",
        help="crawler maintenance verbs (recrawl-check, ...)",
    )
    crawler_sub = crawler_cmd.add_subparsers(dest="crawler_op", required=True)

    recrawl_check_cmd = crawler_sub.add_parser(
        "recrawl-check",
        help=(
            "send conditional HEAD requests for ingested documents and "
            "classify each as fresh/stale/gone/unreachable"
        ),
    )
    recrawl_check_cmd.add_argument(
        "--domain",
        default=None,
        help="restrict to documents whose URI contains this domain",
    )
    recrawl_check_cmd.add_argument(
        "--limit",
        type=int,
        default=100,
        help="check at most N documents (oldest checks first; default: 100)",
    )
    recrawl_check_cmd.set_defaults(func=_cmd_crawler_recrawl_check)

    return p


def main(argv: list[str] | None = None) -> int:
    args = build_parser().parse_args(argv)
    return args.func(args)


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