arborist/aborist/qa/runner.py

"""Q&A runner: cache-first lookup -> inference fallback -> provable record.

Implements the v9.8 admissibility invariant:
    No record reused unless all 8 cache_key dimensions match AND state
    is 'live' (not failed/stale/quarantined).

Cache hit  -> persisted audit_mode (STRICT/HYBRID/UNGROUNDED).
Cache miss -> call ChatClient, run faithfulness check, classify, store
              record, audit event.
"""

from __future__ import annotations

import json
import sqlite3
import time

from aborist import (
    CANONICALIZATION_VERSION,
    SCHEMA_VERSION,
)
from aborist.compress import unpack_chunk
from aborist.merkle import MerkleTree, proof_to_dict
from aborist.qa.client import ChatClient
from aborist.qa.prompts import (
    CLAIM_LATTICE_GROUNDING_REMINDER,
    CLAIM_LATTICE_JSON_GROUNDING_REMINDER,
    CLAIM_LATTICE_JSON_SYSTEM_PROMPT,
    CLAIM_LATTICE_SYSTEM_PROMPT,
)
from aborist.qa.keys import (
    DEFAULT_FIDELITY,
    DEFAULT_QUESTION_DEDUP,
    FIDELITY_MODES,
    QUESTION_DEDUP_MODES,
    cache_key,
    canonical_question,
    conversation_hash,
    governance_policy_hash,
    model_profile_hash,
    question_hash,
    verifier_policy_hash,
)
from aborist.qa.dag import build_run_dag
from aborist.qa.evidence import (
    build_evidence_map,
    evidence_map_root,
    render_evidence_map,
    render_evidence_map_for_json,
)
from aborist.qa.repair import mechanical_repair, reprompt_repair
from aborist.qa.verify import (
    ANSWER_MODES,
    CLAIM_LATTICE_JSON_SCHEMA,
    DEFAULT_ANSWER_MODE,
    verify_claim_lattice,
    verify_claim_lattice_json,
    verify_quotes,
)
from aborist.store import append_audit, transaction

try:
    from aborist.wikitext import BASE_VERSION as _WIKITEXT_BASE_VERSION
    from aborist.wikitext import to_base as _wikitext_to_base
except ImportError:  # pragma: no cover
    _WIKITEXT_BASE_VERSION = None
    _wikitext_to_base = None


DEFAULT_POLICY = {
    # Ticket #000007 — query-layer hyphen-fold marker. See
    # aborist/qa/query.py:DEFAULT_QUERY_POLICY for rationale.
    "hyphen_fold_v1": True,
    # Ticket #000006 amend 2026-05-02b (Rule 9). See
    # aborist/qa/query.py:DEFAULT_QUERY_POLICY for full rationale.
    "claim_lattice_subject_tokens_absent_threshold": 3,
    "system_prompt": (
        "Answer the user's question based ONLY on the document below. "
        "For EVERY factual claim, include a verbatim quote from the "
        "document enclosed in double quotes (\"...\"). The quoted span "
        "must appear word-for-word. Make a claim only when a verbatim "
        "quote in the document directly supports it. "
        "If the answer is in the document, write it. "
        "If the answer is absent from the document, say 'I don't know "
        "based on the provided document.' and stop there. "
        "Stay inside the document at all times."
    ),
    # Restated rule fired as a user message right before the document +
    # question arrive. See aborist/qa/query.py for the rationale (recent
    # user-turn instructions outweigh decayed system-turn rules in 8B
    # instruction-tuned models).
    "grounding_reminder": (
        "REMINDER: wrap every factual claim in double quotes (\"...\") "
        "and the quoted span must appear word-for-word in the "
        "document. Each claim earns a verbatim quote. "
        "Now answer the question on the next message."
    ),
    "temperature": 0.1,
    "top_p": 1.0,
    "max_tokens": 512,
    "entity_policy": "proximity",
    "entity_proximity_n": 3,
    "entity_proximity_window": 300,
    # Mechanical answer repair after first verify. Off by default; see
    # aborist/qa/query.py for semantics.
    "repair_enabled": False,
    "repair_max_reprompts": 0,
    # Strip wikitext markup before the LLM ever sees the context. Lets
    # Hermes quote prose verbatim and shrinks token bills (~43% on
    # Wikipedia chunks). Bumps governance_policy_hash so prior cached
    # answers under raw-wikitext policy stay distinct on lookup. Set
    # via the wikitext extras; no-op if mwparserfromhell isn't installed.
    "base_version": _WIKITEXT_BASE_VERSION,
    # G0 / CTI — claim-lattice-pointer answer mode. "quote" (default):
    # existing behavior, model writes prose with verbatim quotes inline.
    # "claim_lattice_pointer": runtime builds an evidence map and shows
    # the model short pointer ids (E1, E2, …); model writes natural
    # prose with bracket pointer tags ("Claim. [E12]") instead of
    # quoting source text. Renderer interpolates literal spans at
    # display time. Synthetic-elision-by-construction-impossible: the
    # model never types the quote string. Two-layer id discipline keeps
    # the cache & run-DAG keyed on content-addressed evidence_ids.
    # Folds into governance_policy_hash so two modes write under
    # different cache_keys and never alias. No iterative repair in
    # pointer mode (one-shot benchmark discipline).
    "answer_mode": DEFAULT_ANSWER_MODE,
    "claim_lattice_system_prompt": CLAIM_LATTICE_SYSTEM_PROMPT,
    "claim_lattice_grounding_reminder": CLAIM_LATTICE_GROUNDING_REMINDER,
    # Allowed source roles for claim-lattice verification. Roles outside
    # this set get classified SOURCE_ROLE_BLOCKED and downgrade the
    # verdict. Mirrors aborist.qa.verify.DEFAULT_ALLOWED_SOURCE_ROLES;
    # noisy_background_source / sequel_background_source are excluded by
    # default. Folds into governance_policy_hash on change.
    "claim_lattice_allowed_source_roles": [
        "primary_answer_source",
        "secondary_context_source",
        "background_source",
        "unclassified",
        # Self-promoted providence records (`aborist://providence/`
        # URI scheme). Trusted-as-fact substrate per the
        # self-reference design — STRICT live records past the
        # kindergarten window. See
        # docs/self-reference-design.md for the
        # falsification trust model.
        "self_reference_source",
    ],
    # Hard cap on pointer ids per claim line — mirrors prompt Rule 9.
    # Lines exceeding this cap classify as SCHEMA_INVALID and the
    # verdict can no longer reach STRICT. Folds into
    # governance_policy_hash so changing the cap invalidates prior
    # cached records.
    "claim_lattice_max_pointers_per_claim": 2,
    # Minimum claim-token coverage required for the citation-overlap
    # check (Rule 6) to pass. Pre-2026-04-30 the threshold was implicit
    # at "≥1 shared token", which let through lazy-anchored claims
    # whose only overlap was a single topical word (e.g. "Yale
    # University... [E9]" cited to a highway-data span containing only
    # "Connecticut"). 0.30 means a 10-token claim needs ≥3 of its
    # content tokens to appear in the cited span. Short claims (≤3
    # content tokens) keep the old ≥1-token floor so narrow factoids
    # like "Steve Jobs co-founded Apple" still pass. Folds into
    # governance_policy_hash on change.
    "claim_lattice_min_citation_coverage": 0.30,
    # Bare-name claim guard. A claim with fewer than this many content
    # tokens (>=4 chars, post-spotlight-stopword) is rejected as
    # SCHEMA_INVALID. Catches the JP-dinosaurs lazy-anchor where
    # "Triceratops. [E16]" passes lexical overlap on a single token
    # even when E16 is a video-game tie-in chunk rather than the film
    # article. Default 2: bare-entity-name claims (one content token
    # after stopword strip) fail; sentence-shape claims pass. Note
    # ``_content_tokens`` already filters "appears", "shown", etc. so
    # "Trex appears" → 1 content token (filtered), "Trex appears in
    # the film" → 2 content tokens (passes). Folds into
    # governance_policy_hash on change.
    "claim_lattice_min_claim_content_tokens": 2,
    # Lazy-anchor smell auto-demote. When >= threshold of verified
    # pointer-pairs cite a single pointer AND there are >= min_pairs
    # total, cap audit_mode at HYBRID. The smell sidecar was advisory
    # only pre-2026-04-30; now it's load-bearing. STRICT requires
    # diverse anchoring across pointers.
    "claim_lattice_lazy_anchor_demote_threshold": 0.5,
    "claim_lattice_lazy_anchor_demote_min_pairs": 3,
    # Warrant-lite — relation-question hard check (Ticket H from
    # feedback-3, 2026-05-01). Detects relation-shape questions
    # ("who is X's boss?", "who founded Y?") and requires the cited
    # span to contain at least one named answer entity (proper-noun
    # phrase) from the claim. Catches the Homer-Simpson lazy-anchor
    # case where claim asserts "Mr. Burns" but cited span is the
    # voice-actor bio. WARRANT_MISSING violations cap audit_mode
    # at HYBRID. See aborist/qa/warrant.py.
    "claim_lattice_warrant_check_enabled": True,
    "claim_lattice_deflection_check_enabled": True,
    # Format-collapse check (pointer-mode only): when the model emits
    # ≥5 meaningful prose lines with zero `[E\d+]` pointer tags, it
    # abandoned the claim_lattice_pointer protocol entirely. Soft-demote
    # so audit display surfaces "format collapsed" vs "graceful per-
    # claim refusal" — different failure shapes, same UNGROUNDED rung.
    # JSON-mode collapse already shows up as SCHEMA_INVALID so this
    # check is redundant there. Surfaced 2026-05-02 by fox's "Winners
    # of all major sports?" case where Hermes dumped 50+ free-form
    # sentences.
    "claim_lattice_format_collapse_check_enabled": True,
    # Quantifier preflight guard (Ticket #000008 Phase 2). Per-call
    # claim cap derived from the question's quantifier intensity and
    # the configured model profile (aborist/qa/model_profiles.py).
    # Phase 2 lands the lookup wiring with apply_caps=False per
    # §10.11.3 dry-run discipline — claim_cap_applied is computed
    # and reported on the result dict, but the verifier still uses
    # claim_lattice_max_claims_per_answer as the actual cap.
    # Operator flips quantifier_guard_apply_caps=True after dry-run
    # bench review confirms classifier output across the full
    # question set.
    #
    # Six-level disable hierarchy (§10.11.2):
    #   - quantifier_guard_enabled: master kill (False = no
    #     classifier output, no cap lookup, no telemetry).
    #   - quantifier_guard_apply_caps: dry-run gate (True = cap
    #     applied; False = cap reported but not applied).
    #   - quantifier_caps_by_intensity: per-call override dict;
    #     wins over the model_profiles.py table when present.
    #   - quantifier_guard_modes: list of answer_modes the guard
    #     applies to. Quote mode opts out by default — already
    #     stable HYBRID 0.455 on baseline, different failure shape.
    "quantifier_guard_enabled": True,
    "quantifier_guard_apply_caps": False,
    "quantifier_caps_by_intensity": {},
    "quantifier_guard_modes": ["claim_lattice_pointer", "claim_lattice"],
    # Phase 3 — broad-quantifier reminder injection. Default OFF
    # because Hermes-3-8B already ignores parts of the existing
    # reminder under enumeration pressure (ticket §3 Option B con).
    # Operator opts in per-call after Phase 2 dry-run telemetry
    # confirms which broad-shape rows actually need the reminder.
    "quantifier_reminder_enabled": False,
    # Phase 4 — strict reject for broad-unbounded queries. When True
    # AND intensity ∈ {ALL, COMPREHENSIVE, OPEN_REQUEST} AND
    # scope_bound_hint == "unbounded", query()/ask() return UNGROUNDED
    # before the LLM call with a BROAD_QUANTIFIER_REJECTED violation.
    # Saves the ~10-15s LLM call on rejected runs. Default OFF — opt-in
    # via --reject-broad CLI flag or per-call policy override.
    # Bounded universals (e.g. all members of the Beatles, year-anchored
    # questions) are NOT rejected per §10.1.
    "quantifier_reject_broad": False,
    # Claim-count ceiling. Bench finding (2026-04-30 york-england):
    # "tell me all there is to know about X" prompted Hermes to spam
    # 26-59 encyclopedic claims sourced from training, only 2-4 of
    # which grounded in retrieval. Atomic-claim prompt rule (b5925c8)
    # cut this to ~10, but a hard structural cap is defense in depth.
    # Cap of 12 admits typical entity-list questions (5-7 dinosaurs,
    # Simpsons + pets) while flagging the runaway shape. Folds into
    # governance_policy_hash on change.
    "claim_lattice_max_claims_per_answer": 12,
    # JSON variant — `answer_mode="claim_lattice"`. Mirrors the
    # multi-source query path. Pairs with grammar-constrained inference
    # (vLLM guided_json, Claude/GPT-4 native JSON, Qwen 3.6 reasoner).
    # Lenient pre-parser in verify_claim_lattice_json keeps the path
    # survivable on inference paths without grammar guidance.
    "claim_lattice_json_system_prompt": CLAIM_LATTICE_JSON_SYSTEM_PROMPT,
    "claim_lattice_json_grounding_reminder": CLAIM_LATTICE_JSON_GROUNDING_REMINDER,
    "claim_lattice_use_guided_json": True,
    # JSON-mode stop sequences. Hermes-3-8B sometimes spams whitespace
    # / newlines after the closing brace on broad-descriptive shapes
    # ("plot of X", "tell me about Y") — the response runs out the
    # max_tokens budget and the lenient parser sees truncated JSON.
    # Stopping on a blank line cuts the runaway. JSON-mode output
    # never legitimately contains a blank line (single object, single
    # line) so this is a safe filter. Folds into
    # governance_policy_hash on change.
    "claim_lattice_json_stop_sequences": ["\n\n"],
}


def _ms_since(t: float) -> float:
    return round((time.monotonic() - t) * 1000, 1)


def ask(
    conn: sqlite3.Connection,
    *,
    document_root: str,
    question: str,
    client: ChatClient,
    model_id: str,
    revision: str = "",
    quantization: str = "",
    policy: dict | None = None,
    chain: str = "private",
    fidelity: str | None = None,
) -> dict:
    """Look up cached answer or run inference. Returns a result dict.

    See ``aborist.qa.query.query`` for `fidelity` semantics — it
    controls lookup tolerance: ``"strict"`` only checks the cache_key
    matching the call's ``policy["question_dedup"]``; the default
    ``"equivalence_class"`` falls back to the alternate dedup mode's
    cache_key on miss so a fast-cache agent can reuse records written
    under either mode. Result includes ``lookup_path``.
    """
    policy = policy or DEFAULT_POLICY
    if fidelity is None:
        fidelity = policy.get("fidelity", DEFAULT_FIDELITY)
    if fidelity not in FIDELITY_MODES:
        raise ValueError(
            f"fidelity must be one of {FIDELITY_MODES}, got {fidelity!r}"
        )
    # Quantifier preflight (Ticket #000008 Phase 1+2). Same wiring
    # as query() — see aborist/qa/query.py for the rationale and
    # disable hierarchy.
    from aborist.qa.model_profiles import cap_for_intensity
    from aborist.qa.quantifier import classify_question_quantifier
    answer_mode_for_guard = policy.get("answer_mode", "quote")
    quantifier_guard_on = bool(policy.get("quantifier_guard_enabled", True))
    quantifier_guard_modes = policy.get(
        "quantifier_guard_modes",
        ["claim_lattice_pointer", "claim_lattice"],
    )
    quantifier_mode_gated = answer_mode_for_guard in (quantifier_guard_modes or [])
    if quantifier_guard_on:
        quantifier = classify_question_quantifier(question)
    else:
        quantifier = {
            "intensity": None,
            "matched_token": None,
            "explicit_count": None,
            "is_broad": False,
            "operational_shape": None,
            "scope_bound_hint": "unknown",
            "classifier_version": None,
        }
    if quantifier_guard_on and quantifier_mode_gated and quantifier["intensity"]:
        claim_cap_lookup = cap_for_intensity(
            model_profile_id=model_id,
            intensity=quantifier["intensity"],
            explicit_count=quantifier["explicit_count"],
            policy_overrides=policy.get("quantifier_caps_by_intensity") or None,
        )
    else:
        claim_cap_lookup = None
    quantifier_apply_caps = bool(policy.get("quantifier_guard_apply_caps", False))
    _policy_max_claims = int(policy.get("claim_lattice_max_claims_per_answer", 12))
    if quantifier_apply_caps and claim_cap_lookup is not None:
        effective_max_claims = int(claim_cap_lookup)
    else:
        effective_max_claims = _policy_max_claims
    t_start = time.monotonic()

    doc = conn.execute(
        "SELECT document_uri, chunking_version FROM documents "
        "WHERE document_root = ?",
        (document_root,),
    ).fetchone()
    if doc is None:
        return {"status": "unknown_document"}

    chunk_rows = conn.execute(
        "SELECT idx, leaf_hash, content FROM chunks "
        "WHERE document_root = ? ORDER BY idx ASC",
        (document_root,),
    ).fetchall()
    if not chunk_rows:
        return {"status": "unknown_document"}
    if any(r["content"] is None for r in chunk_rows):
        return {"status": "source_cold", "msg": "rehydrate before asking"}

    answer_mode = policy.get("answer_mode", DEFAULT_ANSWER_MODE)
    if answer_mode not in ANSWER_MODES:
        raise ValueError(
            f"policy['answer_mode'] must be one of {ANSWER_MODES}, got {answer_mode!r}"
        )

    chunk_texts = [unpack_chunk(r["content"]) for r in chunk_rows]
    document_text = "\n\n".join(chunk_texts)

    # Wikitext → prose before the LLM sees it. The model can then quote
    # verbatim against the prose form; the verifier compares like-against-
    # like. Idempotent if context is already plain prose. Gated on
    # policy["base_version"] so this is part of governance_policy_hash.
    if policy.get("base_version") and _wikitext_to_base is not None:
        document_text = _wikitext_to_base(document_text)
        chunk_texts = [_wikitext_to_base(t) for t in chunk_texts]

    evidence_map = []
    if answer_mode == "claim_lattice_pointer":
        # Quote-by-pointer: one evidence object per chunk. The model sees
        # the literal spans labeled with content-addressed IDs and is
        # instructed to reference IDs, not type quote text. Synthetic
        # elision is impossible by construction — the model never produces
        # the quote string.
        chunks_for_map = [
            {
                "source_root": document_root,
                "document_uri": doc["document_uri"],
                "title": None,
                "chunk_idx": r["idx"],
                "chunk_root": r["leaf_hash"],
                "span": chunk_texts[i],
                "source_role": "primary_answer_source",
            }
            for i, r in enumerate(chunk_rows)
        ]
        evidence_map = build_evidence_map(chunks_for_map)
        sys_prompt = policy["claim_lattice_system_prompt"]
        grounding_reminder = policy.get("claim_lattice_grounding_reminder")
        rendered_evidence = render_evidence_map(evidence_map)

        def _user_payload(q: str) -> str:
            return f"EVIDENCE:\n\n{rendered_evidence}\n\n---\n\nQUESTION: {q}"
    elif answer_mode == "claim_lattice":
        # JSON variant — same per-chunk evidence map as pointer mode,
        # blocks labeled with content-addressed evidence_id (long hex)
        # since the model emits IDs in JSON. Pairs with grammar-
        # constrained inference; lenient pre-parser handles drift.
        chunks_for_map = [
            {
                "source_root": document_root,
                "document_uri": doc["document_uri"],
                "title": None,
                "chunk_idx": r["idx"],
                "chunk_root": r["leaf_hash"],
                "span": chunk_texts[i],
                "source_role": "primary_answer_source",
            }
            for i, r in enumerate(chunk_rows)
        ]
        evidence_map = build_evidence_map(chunks_for_map)
        sys_prompt = policy.get(
            "claim_lattice_json_system_prompt",
            policy["claim_lattice_system_prompt"],
        )
        grounding_reminder = policy.get(
            "claim_lattice_json_grounding_reminder",
            policy.get("claim_lattice_grounding_reminder"),
        )
        rendered_evidence = render_evidence_map_for_json(evidence_map)

        def _user_payload(q: str) -> str:
            return f"EVIDENCE:\n\n{rendered_evidence}\n\n---\n\nQUESTION: {q}"
    else:
        sys_prompt = policy["system_prompt"]
        grounding_reminder = policy.get("grounding_reminder")

        def _user_payload(q: str) -> str:
            return f"Document:\n\n{document_text}\n\n---\n\nQuestion: {q}"

    # System sets the policy; a user-turn reminder restates the rule one
    # message before the payload arrives. Payload (document or evidence
    # map + question) lands last as the most-recent tokens before
    # generation.
    messages = [{"role": "system", "content": sys_prompt}]
    if grounding_reminder:
        messages.append({"role": "user", "content": grounding_reminder})
    # Quantifier-specific reminder (Ticket #000008 Phase 3, default
    # off). When the question is broad (ALL/COMPREHENSIVE/OPEN_
    # REQUEST) and the operator opted in via
    # quantifier_reminder_enabled=True, append a one-line reminder
    # restating the cap and the no-prior-enumeration rule.
    # quantifier_reminder_enabled defaults to False because Hermes-3-8B
    # already ignores parts of the existing reminder under enumeration
    # pressure (§3 Option B con); empirical effect requires bench
    # measurement before flipping default-on (§10.8 decision tree).
    if (
        quantifier_guard_on
        and quantifier_mode_gated
        and quantifier.get("is_broad")
        and bool(policy.get("quantifier_reminder_enabled", False))
    ):
        from aborist.qa.quantifier_reminder import broad_quantifier_reminder
        broad = broad_quantifier_reminder(
            intensity=quantifier["intensity"],
            cap=effective_max_claims,
            scope_bound_hint=quantifier["scope_bound_hint"],
        )
        if broad:
            messages.append({"role": "user", "content": broad})
    messages.append({"role": "user", "content": _user_payload(question)})

    mhash = model_profile_hash(model_id, revision, quantization)

    # Dedup-mode-aware cache_key. See aborist/qa/query.py for rationale —
    # policy_variant matches the alternate mode so governance_policy_hash
    # agrees with what an agent under that mode would have written,
    # enabling cross-silo fallback.
    def _ckey_for_mode(mode: str) -> str:
        canon_q = canonical_question(question, mode=mode)
        canon_msgs = list(messages[:-1]) + [
            {"role": "user", "content": _user_payload(canon_q)},
        ]
        policy_variant = dict(policy, question_dedup=mode)
        return cache_key(
            document_root,
            question_hash(question, mode=mode),
            mhash,
            conversation_hash(canon_msgs),
            governance_policy_hash(policy_variant),
            SCHEMA_VERSION,
            CANONICALIZATION_VERSION,
            doc["chunking_version"],
            verifier_policy_hash(policy_variant),
        )

    ghash = governance_policy_hash(policy)  # for the legacy INSERT below

    primary_dedup = policy.get("question_dedup", DEFAULT_QUESTION_DEDUP)
    if primary_dedup not in QUESTION_DEDUP_MODES:
        raise ValueError(
            f"policy['question_dedup'] must be one of {QUESTION_DEDUP_MODES}, "
            f"got {primary_dedup!r}"
        )
    # Re-derive the per-mode hashes for use in the INSERT below. _ckey_for_mode
    # already builds them, but the legacy INSERT references qhash/chash by name.
    qhash = question_hash(question, mode=primary_dedup)
    canonical_q_primary = canonical_question(question, mode=primary_dedup)
    canonical_messages_primary = list(messages[:-1]) + [
        {"role": "user", "content": _user_payload(canonical_q_primary)},
    ]
    chash = conversation_hash(canonical_messages_primary)
    primary_ckey = _ckey_for_mode(primary_dedup)
    ckey = primary_ckey  # legacy name for the rest of the function

    t_lookup = time.monotonic()
    cached = conn.execute(
        "SELECT * FROM providence_cache "
        "WHERE cache_key = ? AND falsification_state = 'live'",
        (primary_ckey,),
    ).fetchone()
    hit_ckey = primary_ckey
    lookup_path = primary_dedup if cached is not None else None
    if cached is None and fidelity == "equivalence_class":
        other_mode = (
            "equivalence_class" if primary_dedup == "strict" else "strict"
        )
        other_ckey = _ckey_for_mode(other_mode)
        if other_ckey != primary_ckey:
            cached = conn.execute(
                "SELECT * FROM providence_cache "
                "WHERE cache_key = ? AND falsification_state = 'live'",
                (other_ckey,),
            ).fetchone()
            if cached is not None:
                hit_ckey = other_ckey
                lookup_path = f"{other_mode}_fallback"
    cache_lookup_ms = _ms_since(t_lookup)
    if cached is not None:
        with transaction(conn):
            now = int(time.time())
            conn.execute(
                "UPDATE providence_cache "
                "SET hit_count = hit_count + 1, last_hit_at = ? "
                "WHERE cache_key = ?",
                (now, hit_ckey),
            )
        return {
            "status": "cache_hit",
            "audit_mode": cached["audit_mode"],
            "cache_key": hit_ckey,
            "lookup_path": lookup_path,
            "source_root": document_root,
            "answer_text": cached["answer_text"],
            "merkle_proof": json.loads(cached["merkle_proof"]),
            "n_quotes": cached["n_quotes"],
            "n_verified": cached["n_verified"],
            "verifier_method": cached["verifier_method"],
            "unverified_quotes": (
                json.loads(cached["unverified_quotes"])
                if cached["unverified_quotes"]
                else []
            ),
            "partially_verified_quotes": [],
            # Quantifier preflight (Ticket #000008 Phase 1+2). Pure
            # on the question string, so cache hits re-classify
            # cheaply and carry the same schema as miss-path rows.
            "quantifier_intensity": quantifier["intensity"],
            "quantifier_matched_token": quantifier["matched_token"],
            "scope_bound_hint": quantifier["scope_bound_hint"],
            "quantifier_explicit_count": quantifier["explicit_count"],
            "claim_cap_applied": claim_cap_lookup,
            "timings": {
                "cache_lookup_ms": cache_lookup_ms,
                "llm_ms": None,
                "total_ms": _ms_since(t_start),
            },
        }

    t_llm = time.monotonic()
    # JSON mode: pass guided_json schema so vLLM constrains output at
    # sampling time. Endpoints without guided-decoding silently drop the
    # field; the lenient pre-parser handles whatever drift remains.
    extra_body: dict | None = None
    stop_seqs: list[str] | None = None
    if answer_mode == "claim_lattice" and policy.get(
        "claim_lattice_use_guided_json", True
    ):
        extra_body = {"guided_json": CLAIM_LATTICE_JSON_SCHEMA}
    if answer_mode == "claim_lattice":
        # JSON-mode token-runaway guard. On broad-descriptive /
        # comparison questions Hermes-3-8B sometimes spams whitespace
        # / newlines after the closing brace until max_tokens
        # exhausts; the resulting truncated payload won't parse and
        # the run lands UNGROUNDED 0/0 at 12-15s instead of 2-4s.
        # Stopping on a blank line (\n\n) cuts the runaway —
        # well-formed JSON-mode output never contains a blank line
        # since the model emits a single object on one line (or
        # with simple internal newlines).
        stop_seqs = list(policy.get(
            "claim_lattice_json_stop_sequences", ["\n\n"]
        ))
    raw_answer = client.chat_completion(
        messages,
        model=model_id,
        temperature=policy["temperature"],
        max_tokens=policy["max_tokens"],
        top_p=policy.get("top_p", 1.0),
        extra_body=extra_body,
        stop=stop_seqs,
    )
    llm_ms = _ms_since(t_llm)

    repair_changes: list[dict] = []
    pre_repair_verdict: dict | None = None

    if answer_mode == "claim_lattice_pointer":
        verdict = verify_claim_lattice(
            raw_answer,
            evidence_map,
            allowed_source_roles=tuple(
                policy.get(
                    "claim_lattice_allowed_source_roles",
                    [
                        "primary_answer_source",
                        "secondary_context_source",
                        "background_source",
                        "unclassified",
                    ],
                )
            ),
            max_pointers_per_claim=int(policy.get(
                "claim_lattice_max_pointers_per_claim", 2
            )),
            min_citation_coverage=float(policy.get(
                "claim_lattice_min_citation_coverage", 0.30
            )),
            min_claim_content_tokens=int(policy.get(
                "claim_lattice_min_claim_content_tokens", 3
            )),
            lazy_anchor_demote_threshold=float(policy.get(
                "claim_lattice_lazy_anchor_demote_threshold", 0.5
            )),
            lazy_anchor_demote_min_pairs=int(policy.get(
                "claim_lattice_lazy_anchor_demote_min_pairs", 3
            )),
            max_claims_per_answer=effective_max_claims,
            subject_tokens_absent_threshold=int(policy.get(
                "claim_lattice_subject_tokens_absent_threshold", 3
            )),
            question=question,
            warrant_check_enabled=bool(policy.get(
                "claim_lattice_warrant_check_enabled", True
            )),
            deflection_check_enabled=bool(policy.get(
                "claim_lattice_deflection_check_enabled", True
            )),
            format_collapse_check_enabled=bool(policy.get(
                "claim_lattice_format_collapse_check_enabled", True
            )),
        )
        # Rendered prose (literal spans interpolated) is the user-facing
        # answer text — never the model's raw pointer-line output. If
        # rendering produced nothing (no valid claims), persist the raw
        # output so an operator can see what the model actually said.
        rendered = verdict["rendered_text"]
        answer_text = rendered if rendered else raw_answer
    elif answer_mode == "claim_lattice":
        verdict = verify_claim_lattice_json(
            raw_answer,
            evidence_map,
            allowed_source_roles=tuple(
                policy.get(
                    "claim_lattice_allowed_source_roles",
                    [
                        "primary_answer_source",
                        "secondary_context_source",
                        "background_source",
                        "unclassified",
                    ],
                )
            ),
            max_evidence_per_claim=int(policy.get(
                "claim_lattice_max_pointers_per_claim", 2
            )),
            min_citation_coverage=float(policy.get(
                "claim_lattice_min_citation_coverage", 0.30
            )),
            max_claims_per_answer=effective_max_claims,
            subject_tokens_absent_threshold=int(policy.get(
                "claim_lattice_subject_tokens_absent_threshold", 3
            )),
            question=question,
            warrant_check_enabled=bool(policy.get(
                "claim_lattice_warrant_check_enabled", True
            )),
            deflection_check_enabled=bool(policy.get(
                "claim_lattice_deflection_check_enabled", True
            )),
        )
        rendered = verdict["rendered_text"]
        answer_text = rendered if rendered else raw_answer
    else:
        answer_text = raw_answer
        verdict = verify_quotes(
            answer_text,
            document_text,
            entity_policy=policy.get("entity_policy", "hybrid"),
            proximity_n=policy.get("entity_proximity_n", 3),
            proximity_window=policy.get("entity_proximity_window", 300),
        )

        def _verify(text: str) -> dict:
            return verify_quotes(
                text,
                document_text,
                entity_policy=policy.get("entity_policy", "hybrid"),
                proximity_n=policy.get("entity_proximity_n", 3),
                proximity_window=policy.get("entity_proximity_window", 300),
            )

        if (
            policy.get("repair_enabled")
            and verdict["audit_mode"] != "STRICT"
            and verdict.get("unverified_quotes")
        ):
            repair_result = mechanical_repair(
                answer_text, verdict["unverified_quotes"], document_text
            )
            if repair_result["changes"]:
                new_verdict = _verify(repair_result["repaired_text"])
                if new_verdict["n_verified"] >= verdict["n_verified"]:
                    pre_repair_verdict = verdict
                    answer_text = repair_result["repaired_text"]
                    verdict = new_verdict
                    repair_changes = list(repair_result["changes"])

            max_reprompts = int(policy.get("repair_max_reprompts", 0))
            for _ in range(max_reprompts):
                if (
                    verdict["audit_mode"] == "STRICT"
                    or not verdict.get("unverified_quotes")
                ):
                    break
                new_text = reprompt_repair(
                    chat_client=client,
                    model_id=model_id,
                    original_messages=messages,
                    original_answer=answer_text,
                    failed_quotes=verdict["unverified_quotes"],
                    policy=policy,
                )
                if not new_text:
                    break
                new_verdict = _verify(new_text)
                if new_verdict["n_verified"] > verdict["n_verified"]:
                    if pre_repair_verdict is None:
                        pre_repair_verdict = verdict
                    answer_text = new_text
                    verdict = new_verdict
                    repair_changes.append({
                        "action": "reprompt_rewrite",
                        "diagnosis": "model_feedback_loop",
                    })
                else:
                    break

    unverified_blob = (
        json.dumps(verdict["unverified_quotes"], separators=(",", ":"))
        if verdict["unverified_quotes"]
        else None
    )

    leaves = [bytes.fromhex(r["leaf_hash"]) for r in chunk_rows]
    tree = MerkleTree.build(leaves)
    proof_obj = {
        "document_root": document_root,
        "chunk_0_proof": proof_to_dict(tree.proof(0)),
    }
    proof_blob = json.dumps(proof_obj, separators=(",", ":"))

    # Per-run Merkle-DAG (see aborist/qa/dag.py). Single-doc shape: the
    # only "source" is document_root. Pointer mode swaps the 7-stage
    # quote shape for the 9-stage CTI shape — context drops out and
    # answer splits into raw_answer / parsed_claim_lattice / render.
    ev_root = evidence_map_root(evidence_map) if evidence_map else None
    parsed_lattice = None
    is_lattice_mode = answer_mode in ("claim_lattice_pointer", "claim_lattice")
    if is_lattice_mode:
        # Per-claim list of {claim_text, content-addressed evidence_ids}
        # for the parsed_claim_lattice node hash. Pointer ids are
        # run-dependent; evidence_ids are content-addressed → the run-
        # DAG hashes the run-stable form. Same shape for JSON and
        # pointer; verifier already returns evidence_id_pairs.
        evidence_id_pairs = verdict.get("evidence_id_pairs") or []
        parsed_lattice = [
            {
                "claim_text": cs.get("text", ""),
                "evidence_ids": evidence_id_pairs[i] if i < len(evidence_id_pairs) else [],
            }
            for i, cs in enumerate(verdict.get("claim_statuses") or [])
        ]
    run_dag = build_run_dag(
        question_hash=qhash,
        sources=[{
            "document_root": document_root,
            "source_role": "primary_answer_source",
            "score": None,
            "chunk_idx": None,
        }],
        context_root=document_root,
        conversation_hash=chash,
        answer_text=answer_text,
        audit_mode=verdict["audit_mode"],
        verifier_method=verdict["verifier_method"],
        n_quotes=verdict["n_quotes"],
        n_verified=verdict["n_verified"],
        claim_statuses=verdict.get("claim_statuses", []),
        lookup_path="miss",
        evidence_map_root=ev_root,
        answer_mode=answer_mode if answer_mode != "quote" else None,
        violations=verdict.get("violations"),
        raw_answer_text=raw_answer if is_lattice_mode else None,
        parsed_lattice=parsed_lattice,
        rendered_text=answer_text if is_lattice_mode else None,
    )
    run_dag_blob = json.dumps(run_dag, separators=(",", ":"))

    now = int(time.time())
    with transaction(conn):
        if repair_changes and pre_repair_verdict is not None:
            append_audit(
                conn,
                event_type="providence_repair",
                subject_root=ckey,
                body={
                    "kind": "mechanical",
                    "n_changes": len(repair_changes),
                    "changes": repair_changes,
                    "pre_audit_mode": pre_repair_verdict["audit_mode"],
                    "post_audit_mode": verdict["audit_mode"],
                    "pre_n_verified": pre_repair_verdict["n_verified"],
                    "post_n_verified": verdict["n_verified"],
                },
                ts=now,
            )
        event_hash = append_audit(
            conn,
            event_type="providence_write",
            subject_root=ckey,
            body={
                "source_root": document_root,
                "model_id": model_id,
                "revision": revision,
                "quantization": quantization,
                "chunks_in_context": len(chunk_rows),
                "answer_chars": len(answer_text),
                "audit_mode": verdict["audit_mode"],
                "n_quotes": verdict["n_quotes"],
                "n_verified": verdict["n_verified"],
                "verifier_method": verdict["verifier_method"],
            },
            ts=now,
        )
        conn.execute(
            "INSERT INTO providence_cache "
            "(cache_key, source_root, document_uri, question_hash, question_text, "
            " answer_text, merkle_proof, model_profile_hash, conversation_hash, "
            " governance_policy_hash, schema_version, canonicalization_version, "
            " chunking_version, falsification_state, chain, audit_event_hash, "
            " created_at, hit_count, audit_mode, n_quotes, n_verified, "
            " unverified_quotes, verifier_method, run_dag_root, run_dag_blob) "
            "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'live', ?, ?, ?, 0, "
            " ?, ?, ?, ?, ?, ?, ?)",
            (
                ckey,
                document_root,
                doc["document_uri"],
                qhash,
                question,
                answer_text,
                proof_blob,
                mhash,
                chash,
                ghash,
                SCHEMA_VERSION,
                CANONICALIZATION_VERSION,
                doc["chunking_version"],
                chain,
                event_hash,
                now,
                verdict["audit_mode"],
                verdict["n_quotes"],
                verdict["n_verified"],
                unverified_blob,
                verdict["verifier_method"],
                run_dag["root"],
                run_dag_blob,
            ),
        )

    from aborist.qa.dag import localize_failure as _localize
    failure_stage = _localize(
        audit_mode=verdict["audit_mode"],
        n_sources=1,  # ask() runs against one document
        n_quotes=verdict["n_quotes"],
        n_verified=verdict["n_verified"],
    )
    return {
        "status": "cache_miss_then_written",
        "audit_mode": verdict["audit_mode"],
        "cache_key": ckey,
        "run_dag_root": run_dag["root"],
        "lookup_path": "miss",
        "failure_stage": failure_stage,
        "repair_changes": repair_changes,
        "pre_repair_audit_mode": (
            pre_repair_verdict["audit_mode"] if pre_repair_verdict else None
        ),
        "source_root": document_root,
        "answer_text": answer_text,
        "merkle_proof": proof_obj,
        "n_quotes": verdict["n_quotes"],
        "n_verified": verdict["n_verified"],
        "verifier_method": verdict["verifier_method"],
        "unverified_quotes": verdict["unverified_quotes"],
        "partially_verified_quotes": verdict.get("partially_verified_quotes") or [],
        # Quantifier preflight (Ticket #000008 Phase 1+2). See query.py
        # for full rationale; runner.ask carries the same schema for
        # CLI-side `aborist ask` parity with `aborist query`.
        "quantifier_intensity": quantifier["intensity"],
        "quantifier_matched_token": quantifier["matched_token"],
        "scope_bound_hint": quantifier["scope_bound_hint"],
        "quantifier_explicit_count": quantifier["explicit_count"],
        "claim_cap_applied": claim_cap_lookup,
        # Sidecar smell signals (claim_lattice mode only) — render-
        # layer; never persisted, never in run_dag_root.
        "pointer_id_distribution": verdict.get("pointer_id_distribution"),
        "lazy_anchor_ratio": verdict.get("lazy_anchor_ratio"),
        "timings": {
            "cache_lookup_ms": cache_lookup_ms,
            "llm_ms": llm_ms,
            "total_ms": _ms_since(t_start),
        },
    }