arborist/arborist/qa/providence_query.py

"""Cache-aware claim-lattice query: providence_query wraps run_query
with providence_cache lookup + persist (Phase 2 of #53).

What this does today (minimum-viable):
  - Build the 8-dim cache_key from question + policy + model_id
  - Lookup providence_cache → return cached row on hit
  - Miss: call run_query(corpus, question, chat_client, policy=...) and
    persist the result with audit chain append
  - Return a dict shaped like the legacy query() result so callers
    can swap with minimal changes

What this does NOT do yet (deferred to later Phase 2 sub-steps):
  - Pre-retrieval gates (canonical_projection, crosslang, quantifier,
    metacog, soft_preflight, frame_detection) — they wrap providence_query
    in legacy query(); for now they're skipped, meaning broad questions
    bypass the broad-quantifier guard and crosslang queries bypass the
    sandwich translation edges
  - equivalence_class fallback lookup — the primary cache_key is the
    only one checked
  - Post-retrieval add-ons (answerability, repair, witness, sandwich
    edge-out)
  - retrieval_keywords, translator, extra_body — legacy-only knobs

Legacy `arborist.qa.query.query()` stays the user-facing entry point
until this is bench-proven on the smoke fixture. When parity holds,
query() becomes a thin adapter that delegates here.
"""
from __future__ import annotations

import sqlite3 as _sqlite
import time as _time
from pathlib import Path
from typing import Any

import json as _json

from arborist.qa.client import ChatClient
from arborist.qa.corpus import Corpus
from arborist.qa.corpus_query import run_query
from arborist.qa.dag import build_run_dag
from arborist.qa.keys import (
    cache_key,
    conversation_hash,
    governance_policy_hash,
    model_profile_hash,
    question_hash,
)
from arborist.qa.prompts import (
    CLAIM_LATTICE_GROUNDING_REMINDER,
    CLAIM_LATTICE_SYSTEM_PROMPT,
)
from arborist.store import append_audit


# Schema dimensions for the 8-dim cache_key. Match legacy
# DEFAULT_QUERY_POLICY values; bumping any of these here without
# bumping the source orphans cache records, so always pull from
# the canonical source when callers don't override.
_DEFAULT_SCHEMA_VERSION = "v9.8.0"
_DEFAULT_CANONICALIZATION_VERSION = "norm-v1"
_DEFAULT_CHUNKING_VERSION = "tok-512-v1"


def _build_canonical_messages_for_hash(
    question: str, evidence_text: str
) -> list[dict]:
    """Mirror the message shape run_query passes to chat_completion.

    Must stay in sync with arborist/qa/corpus_query.py:run_query —
    same system prompt, same user-payload template. The byte-identity
    test (tests/test_run_query_byte_identity.py) catches drift.
    """
    user = (
        f"EVIDENCE:\n\n{evidence_text}\n\n"
        f"QUESTION: {question}\n\n"
        f"{CLAIM_LATTICE_GROUNDING_REMINDER}"
    )
    return [
        {"role": "system", "content": CLAIM_LATTICE_SYSTEM_PROMPT},
        {"role": "user", "content": user},
    ]


def _compute_cache_key(
    *,
    question: str,
    evidence_text: str,
    policy: dict,
    model_id: str,
    source_root: str,
    dedup_mode: str = "equivalence_class",
) -> str:
    """Build the 8-dim cache_key for this call.

    Notes on the dimensions:
      - source_root: the Merkle context_root over retrieval sources.
        Caller computes from corpus.snapshot_root() or per-hit
        context_root logic; we accept it as input.
      - evidence_text: what got assembled into the user prompt.
        Critical input to conversation_hash — different evidence →
        different cache_key, even for the same question.
      - policy: full policy dict (NOT the verifier-subset). Same
        canonicalization as legacy.
    """
    return cache_key(
        source_root=source_root,
        question_hash_value=question_hash(question, mode=dedup_mode),
        model_profile_hash_value=model_profile_hash(model_id),
        conversation_hash_value=conversation_hash(
            _build_canonical_messages_for_hash(question, evidence_text)
        ),
        governance_policy_hash_value=governance_policy_hash(policy),
        schema_version=policy.get("schema_version", _DEFAULT_SCHEMA_VERSION),
        canonicalization_version=policy.get(
            "canonicalization_version", _DEFAULT_CANONICALIZATION_VERSION
        ),
        chunking_version=policy.get(
            "chunking_version", _DEFAULT_CHUNKING_VERSION
        ),
    )


def _qa_conn(qa_db: Path | str) -> _sqlite.Connection:
    """Open qa.db read/write. Caller closes."""
    conn = _sqlite.connect(str(qa_db))
    conn.row_factory = _sqlite.Row
    return conn


def _lookup(qa_conn: _sqlite.Connection, ckey: str) -> _sqlite.Row | None:
    """Read one live providence_cache row by cache_key."""
    return qa_conn.execute(
        "SELECT * FROM providence_cache "
        "WHERE cache_key = ? AND falsification_state = 'live'",
        (ckey,),
    ).fetchone()


def _bump_hit(qa_conn: _sqlite.Connection, ckey: str) -> None:
    """Increment hit_count + last_hit_at on a cache row."""
    now = int(_time.time())
    qa_conn.execute(
        "UPDATE providence_cache "
        "SET hit_count = hit_count + 1, last_hit_at = ? "
        "WHERE cache_key = ?",
        (now, ckey),
    )
    qa_conn.commit()


def _burn(qa_conn: _sqlite.Connection, ckey: str) -> int:
    """Delete a live cache row. Returns 1 if a row went away."""
    cur = qa_conn.execute(
        "DELETE FROM providence_cache "
        "WHERE cache_key = ? AND falsification_state = 'live'",
        (ckey,),
    )
    qa_conn.commit()
    return cur.rowcount


def _persist_miss(
    qa_conn: _sqlite.Connection,
    *,
    ckey: str,
    question: str,
    run_result: dict,
    sources: list,
    proof_blob: str,
    source_root: str,
    qhash: str,
    mhash: str,
    chash: str,
    ghash: str,
    schema_version: str,
    canonicalization_version: str,
    chunking_version: str,
    chain: str = "private",
) -> tuple[str, dict]:
    """Write one providence_cache row + one audit event for a fresh
    (miss) result. Returns (audit_event_hash, run_dag_dict).

    ``proof_blob`` is a JSON-serialized {context_root, sources,
    retrieval_purity} object — same shape legacy query() writes, and
    what arborist-viz reads to synthesize the leaf set under a
    multi-source context root.

    Audit chain append happens BEFORE the providence_cache INSERT so
    event_hash is available to set audit_event_hash on the row.
    Both run inside one transaction so a crash mid-way leaves the
    chain consistent.
    """
    audit_mode = run_result.get("audit_mode") or "UNGROUNDED"
    answer_text = run_result.get("answer_text") or ""
    verifier_method = run_result.get("verifier_method") or "none"
    n_quotes = int(run_result.get("n_quotes") or 0)
    n_verified = int(run_result.get("n_verified") or 0)
    violations = run_result.get("violations") or []
    unverified_blob = _json.dumps(
        run_result.get("unverified_quotes") or [], separators=(",", ":")
    )
    raw_answer = run_result.get("raw_answer") or ""
    # Reconstruct context_root from per-source document_roots. Legacy
    # query() uses _context_root which is a Merkle root over the
    # sorted source_roots; reuse that helper for byte-identical roots
    # when sources match.
    from arborist.qa.query import _context_root
    context_root = _context_root([s.get("document_root", "") for s in sources])
    run_dag = build_run_dag(
        question_hash=qhash,
        sources=sources,
        context_root=context_root,
        conversation_hash=chash,
        answer_text=answer_text,
        audit_mode=audit_mode,
        verifier_method=verifier_method,
        n_quotes=n_quotes,
        n_verified=n_verified,
        violations=violations,
        raw_answer_text=raw_answer or None,
    )
    run_dag_blob = _json.dumps(run_dag, separators=(",", ":"))
    now = int(_time.time())
    qa_conn.execute("BEGIN IMMEDIATE")
    try:
        event_hash = append_audit(
            qa_conn,
            event_type="providence_write",
            body={
                "cache_key": ckey,
                "audit_mode": audit_mode,
                "run_dag_root": run_dag["root"],
                "n_quotes": n_quotes,
                "n_verified": n_verified,
            },
            subject_root=source_root,
            ts=now,
        )
        qa_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, ?, ?, ?, ?, ?, ?, ?) "
            "ON CONFLICT(cache_key) DO NOTHING",
            (
                ckey, source_root, "corpus://multi-source", qhash,
                question, answer_text, proof_blob,
                mhash, chash, ghash,
                schema_version, canonicalization_version, chunking_version,
                chain, event_hash, now,
                audit_mode, n_quotes, n_verified,
                unverified_blob, verifier_method,
                run_dag["root"], run_dag_blob,
            ),
        )
        qa_conn.execute("COMMIT")
    except BaseException:
        qa_conn.execute("ROLLBACK")
        raise
    return event_hash, run_dag


def providence_query(
    corpus: Corpus,
    question: str,
    chat_client: ChatClient,
    *,
    qa_db: Path | str,
    policy: dict | None = None,
    model_id: str = "stub",
    burn_existing: bool = False,
    top_k: int = 4,
    max_context_chars: int = 24_000,
    temperature: float = 0.1,
    max_tokens: int = 512,
    progress: Any = None,
) -> dict[str, Any]:
    """Cache-aware claim-lattice query.

    Result shape mirrors run_query but adds cache-status fields:
      status: "cache_hit" | "fresh" | "burned"
      cache_key: the 8-dim key used for lookup + persist
      lookup_path: "primary" (no fallback logic in this minimal version)
      burned_existing: 0 or 1

    Pre/post wrappers (canonical_projection / crosslang / quantifier /
    metacog / answerability / repair / witness) are NOT applied here
    — legacy query() keeps them for now. providence_query is the
    cache-aware retrieval+verify orchestrator; wrap it externally
    when you want those gates.
    """
    policy = policy or {}
    t_total = _time.time()
    burned_existing = 0

    # Disabled no-op shim when caller doesn't pass a progress object —
    # keeps emit-or-not logic local to the call sites.
    if progress is None:
        class _NoProgress:
            def emit(self, *a, **kw): pass
        progress = _NoProgress()

    # 1. RETRIEVAL ONLY — fast phase. Body + title FTS5 in parallel,
    # merge by best bm25 per doc_root. Body alone misses canonical
    # primary articles when high-DF tokens dilute the match (e.g.
    # "when did sesame street first air" → body progressive-AND
    # surfaces AC textbook chunks because "first"/"air" match too
    # broadly; title FTS5 nails "Sesame Street" article at rank 1).
    # The merge picks whichever route ranked a doc best. Title-boost
    # then runs on the union.
    from arborist.qa.corpus import NotSupportedError, apply_title_boost
    progress.emit("search.start", top_k=top_k)
    t_ret = _time.time()

    # Retrieval-side keyword augmentation: when ``policy[
    # "retrieval_keywords"]`` is set (free-text), append to the question
    # passed into FTS5 routes. Verifier-blind, never folded into
    # cache_key (mirrors the legacy ``--retrieval-keywords`` flag,
    # see #000001). Used by ``arborist session`` to flow ancestor
    # cited-titles down a branch so pronoun follow-ups ("who created
    # him?") still surface the prior turn's primary source.
    retrieval_kw = (policy.get("retrieval_keywords") or "").strip()
    retrieval_question = (
        f"{question} {retrieval_kw}" if retrieval_kw else question
    )

    def _safe(method: str):
        try:
            fn = getattr(corpus, method, None)
            if fn is None:
                return []
            return fn(retrieval_question, limit=top_k * 4)
        except NotSupportedError:
            return []

    # Sequential: MultiShardSqliteCorpus.fts_body / fts_title already
    # fan out across shards internally with a ThreadPoolExecutor. An
    # outer pool here would double-parallelize, and single-shard
    # SqliteShardCorpus connections are not thread-safe by default
    # (check_same_thread=True). Run body then title.
    body_hits = _safe("fts_body")
    title_hits = _safe("fts_title")

    # Merge by document_root, keeping the best bm25 score across
    # routes. FTS5 bm25 is "more negative = better" — title typically
    # ranges -10..-20; body can be positive when progressive-AND
    # collapsed to a single weak token. min() picks the better.
    best: dict = {}
    for h in body_hits + title_hits:
        cur = best.get(h.document_root)
        if cur is None or h.score < cur.score:
            best[h.document_root] = h
    hits = sorted(best.values(), key=lambda h: h.score)

    hits = apply_title_boost(
        hits, retrieval_question,
        higher_is_better=getattr(corpus, "higher_is_better", False),
    )[:top_k]
    retrieval_s = _time.time() - t_ret
    progress.emit(
        "search.done", hits=len(hits),
        body=len(body_hits), title=len(title_hits),
        ms=int(retrieval_s * 1000),
    )

    # Provisional sources shape for cache_key + result. The richer
    # form (with used/used_pointer_ids) gets stamped on the miss
    # path after the LLM citations resolve.
    sources = [
        {"document_root": h.document_root,
         "document_uri": h.document_uri,
         "title": h.title,
         "score": h.score}
        for h in hits
    ]
    # Evidence-shape proxy for conversation_hash. The proxy is
    # `title|doc_root[:16]` per source, NOT the full evidence text
    # (legacy convention — see arborist.qa.query._context_root
    # pattern). This means a chunk-content-tweak that doesn't
    # change WHICH docs surfaced won't rotate cache_key — only
    # changes in the doc set do. That's the right cache identity
    # for retrieval-driven Q&A.
    evidence_repr = "\n".join(
        f"{s.get('title','')}|{s.get('document_root','')[:16]}"
        for s in sources
    )
    # source_root is a Merkle root over the RETRIEVED document_roots
    # (legacy convention via arborist.qa.query._context_root) — NOT
    # corpus.snapshot_root() which walks every doc in every shard
    # (~4M rows on the genesis corpus, 40-70s per call). The cache
    # identity question is "which docs did this query surface?", not
    # "what's the global corpus hash" — so context_root is the right
    # dimension.
    from arborist.qa.query import _context_root
    source_root = _context_root([h.document_root for h in hits])
    # Pre-compute each dimension once — needed both for cache_key
    # AND for the persist path below.
    qhash = question_hash(question, mode="equivalence_class")
    mhash = model_profile_hash(model_id)
    chash = conversation_hash(
        _build_canonical_messages_for_hash(question, evidence_repr)
    )
    ghash = governance_policy_hash(policy)
    schema_version = policy.get("schema_version", _DEFAULT_SCHEMA_VERSION)
    canonicalization_version = policy.get(
        "canonicalization_version", _DEFAULT_CANONICALIZATION_VERSION
    )
    chunking_version = policy.get(
        "chunking_version", _DEFAULT_CHUNKING_VERSION
    )
    ckey = cache_key(
        source_root=source_root,
        question_hash_value=qhash,
        model_profile_hash_value=mhash,
        conversation_hash_value=chash,
        governance_policy_hash_value=ghash,
        schema_version=schema_version,
        canonicalization_version=canonicalization_version,
        chunking_version=chunking_version,
    )

    # 4. Cache lookup. Burn first if requested (so a re-lookup after
    # burn correctly misses and re-runs the LLM in the SAME call —
    # legacy convention).
    progress.emit("cache.lookup", primary=ckey[:8])
    qa_conn = _qa_conn(qa_db)
    audit_event_hash = None
    run_dag_root = None
    try:
        if burn_existing:
            burned_existing = _burn(qa_conn, ckey)
            progress.emit("cache.burn", removed=burned_existing)
        t_lookup = _time.time()
        cached = _lookup(qa_conn, ckey)
        if cached is not None and not burn_existing:
            _bump_hit(qa_conn, ckey)
            progress.emit(
                "cache.hit", lookup_path="primary",
                ms=int((_time.time() - t_lookup) * 1000),
            )
            # Cache hit — return cached answer WITHOUT running the
            # LLM. retrieval already happened (we needed it for the
            # cache_key); skipping LLM is the real win.
            from arborist.qa.query import _context_root
            return {
                "status": "cache_hit",
                "cache_key": ckey,
                "lookup_path": "primary",
                "audit_mode": cached["audit_mode"],
                "answer_text": cached["answer_text"],
                "raw_answer": cached["answer_text"],
                "sources": sources,
                "audit_event_hash": cached["audit_event_hash"],
                "run_dag_root": cached["run_dag_root"],
                "verifier_method": cached["verifier_method"] or "none",
                "n_quotes": int(cached["n_quotes"] or 0),
                "n_verified": int(cached["n_verified"] or 0),
                "unverified_quotes": _json.loads(
                    cached["unverified_quotes"] or "[]"
                ),
                "partially_verified_quotes": [],
                "warrant_proven_claim_idxs": [],
                "violations": [],
                "format_collapsed": None,
                "context_root": _context_root(
                    [s.get("document_root", "") for s in sources]
                ),
                "prompt_chars": {
                    "messages_total": 0, "system_prompt": 0,
                    "evidence_or_context": 0, "user_question": 0,
                    "grounding_reminder": 0,
                },
                "answer_chars": len(cached["answer_text"] or ""),
                "timings": _merge_timings(
                    {"search": retrieval_s, "total": _time.time() - t_total},
                    cache_lookup_s=round(_time.time() - t_total - retrieval_s, 3),
                    cache_persist_s=0.0,
                ),
                "burned_existing": 0,
                "elapsed_s": round(_time.time() - t_total, 3),
            }
        # 5. Miss — call run_query for LLM + verify. Pass our hits
        # through so run_query doesn't redo retrieval and discard our
        # title-aware union (previously dropped Paris off the "capitol
        # of paris" query because run_query re-did body-only retrieval
        # internally and high-DF body noise outranked the canonical
        # title-hit primary).
        progress.emit("cache.miss", lookup_path="primary")
        progress.emit("llm.start", model=model_id, top_k=top_k)
        t_llm = _time.time()
        run_result = run_query(
            corpus, question, chat_client,
            model_id=model_id, top_k=top_k,
            max_context_chars=max_context_chars,
            temperature=temperature, max_tokens=max_tokens,
            policy=policy,
            precomputed_hits=hits,
        )
        progress.emit(
            "llm.done",
            audit_mode=run_result.get("audit_mode"),
            n_verified=run_result.get("n_verified"),
            n_quotes=run_result.get("n_quotes"),
            ms=int((_time.time() - t_llm) * 1000),
        )
        # Use the LLM-run sources (with used/used_pointer_ids) for
        # the persist path; fall back to provisional if absent.
        sources = run_result.get("sources") or sources
        # Enrich each source with hit-level retrieval metadata so the
        # persisted merkle_proof matches legacy query()'s shape.
        # Arborist-viz reads `merkle_proof.sources` to synthesize the
        # leaf set under the context_root — an empty list (the prior
        # "[]" placeholder) showed 0 leaves in the lattice.
        hit_by_root = {h.document_root: h for h in hits}
        for s in sources:
            h = hit_by_root.get(s.get("document_root"))
            if h is None:
                continue
            s.setdefault("score", float(h.score))
            s.setdefault("shard", Path(h.shard_id).name if h.shard_id else "")
        # Retrieval-purity sidecar (legacy parity — proof_obj has it).
        # Sidecar signal, not commitment: lets bench scripts spot
        # creeping retrieval-quality regressions without touching the
        # audit chain (the audit body excludes merkle_proof).
        noisy_roles = {"noisy_background_source", "sequel_background_source"}
        primary_rank = next(
            (i for i, s in enumerate(sources, start=1)
             if s.get("source_role") == "primary_answer_source"),
            0,
        )
        noise_sources = [s for s in sources if s.get("source_role") in noisy_roles]
        retrieval_purity = {
            "primary_rank": primary_rank,
            "primary_used": (
                primary_rank > 0
                and sources[primary_rank - 1].get("used") is True
            ),
            "noise_sources_count": len(noise_sources),
            "noise_sources_used": sum(1 for s in noise_sources if s.get("used")),
            "total_sources": len(sources),
            "used_sources": sum(1 for s in sources if s.get("used")),
        }
        proof_obj = {
            "context_root": source_root,
            "sources": sources,
            "retrieval_purity": retrieval_purity,
        }
        proof_blob = _json.dumps(proof_obj, separators=(",", ":"))
        progress.emit("persist.start")
        t_persist = _time.time()
        audit_event_hash, run_dag = _persist_miss(
            qa_conn,
            ckey=ckey, question=question, run_result=run_result,
            sources=sources, proof_blob=proof_blob, source_root=source_root,
            qhash=qhash, mhash=mhash, chash=chash, ghash=ghash,
            schema_version=schema_version,
            canonicalization_version=canonicalization_version,
            chunking_version=chunking_version,
        )
        run_dag_root = run_dag["root"]
        progress.emit(
            "persist.done",
            ms=int((_time.time() - t_persist) * 1000),
        )
    finally:
        qa_conn.close()

    # Build the renderer-shaped result. The render layer in cli.py
    # (`_render_query_human`) and the unfirehose journal emitter
    # (`_emit_query_journal`) read many top-level fields legacy
    # query() emitted; flatten run_query's nested `capacity` +
    # surface every verifier sidecar so the CLI output matches
    # legacy's information density. Sidecars that depend on pre/post
    # gates not yet ported into providence_query (answerability,
    # quantifier_*, frame_detection, soft_preflight_hint,
    # question_state, etc.) stay absent — the renderer handles
    # missing fields gracefully (defaults to blank/zero), and the
    # CLI shows the basics legacy users expect.
    from arborist.qa.query import _context_root
    out = dict(run_result)
    cap = out.get("capacity") or {}
    out.update({
        "status": "burned" if burned_existing else "fresh_persisted",
        "cache_key": ckey,
        "lookup_path": "miss",
        "burned_existing": burned_existing,
        "audit_event_hash": audit_event_hash,
        "run_dag_root": run_dag_root,
        "elapsed_s": round(_time.time() - t_total, 3),
        # Legacy-shape prompt_chars is a DICT (the renderer reads
        # .get('messages_total'), .get('system_prompt'), etc.) —
        # map run_query's flat capacity counts to that schema.
        # grounding_reminder is computed as the remainder so totals
        # reconcile (sys + evidence + user + reminder == total).
        "prompt_chars": _prompt_chars_legacy_shape(cap),
        "answer_chars": cap.get("answer_chars", 0),
        # Context-root over sources (Merkle root of doc_roots).
        "context_root": _context_root(
            [s.get("document_root", "") for s in sources]
        ),
        # Verifier sidecars that run_query already computed but
        # nested under different shapes. Default empty when absent.
        "unverified_quotes": run_result.get("unverified_quotes") or [],
        "partially_verified_quotes": run_result.get(
            "partially_verified_quotes"
        ) or [],
        "warrant_proven_claim_idxs": run_result.get(
            "warrant_proven_claim_idxs"
        ) or [],
        "format_collapsed": run_result.get("format_collapsed"),
        # Per-phase timings: surface the run_query phase dict AS-IS
        # plus a synthetic legacy-shape entry the renderer hooks.
        # Legacy expects: cache_lookup, retrieval, prompt_build,
        # llm, verify, persist. run_query emits: search, context,
        # llm, verify, total. Map best-fit so the renderer's
        # phase-loop prints something for each known key.
        "timings": _merge_timings(
            run_result.get("timings") or {},
            cache_lookup_s=0.0,        # negligible — single SELECT
            cache_persist_s=0.0,       # baked into total today
        ),
    })
    return out


def _prompt_chars_legacy_shape(cap: dict) -> dict:
    """Reshape run_query's flat capacity counts into the dict legacy
    query() emits (and the CLI renderer reads via .get('messages_total'),
    .get('system_prompt'), etc.). grounding_reminder is the remainder
    so the four parts sum to messages_total exactly."""
    total = cap.get("prompt_chars", 0)
    sys_p = cap.get("sys_prompt_chars", 0)
    evidence = cap.get("evidence_chars", 0)
    user = cap.get("question_chars", 0)
    reminder = max(0, total - sys_p - evidence - user)
    return {
        "messages_total": total,
        "system_prompt": sys_p,
        "evidence_or_context": evidence,
        "user_question": user,
        "grounding_reminder": reminder,
    }


def _merge_timings(rq: dict, *, cache_lookup_s: float, cache_persist_s: float) -> dict:
    """Map run_query's per-phase timings + providence's cache phases
    into a renderer-friendly shape.

    The CLI renderer (arborist.cli._render_query_human) reads INT
    millisecond keys (cache_lookup_ms, search_ms, context_ms,
    llm_ms, persist_ms, total_ms). Bench scripts read float-second
    keys (search, context, llm, verify, total). Emit BOTH so neither
    consumer breaks; JSON callers see the full picture either way."""
    search_s = rq.get("search", 0.0)
    context_s = rq.get("context", 0.0)
    llm_s = rq.get("llm", 0.0)
    verify_s = rq.get("verify", 0.0)
    total_s = rq.get("total", 0.0)
    return {
        # Native run_query phase keys (float seconds — bench scripts)
        "search": search_s,
        "context": context_s,
        "llm": llm_s,
        "verify": verify_s,
        "total": total_s,
        # Legacy aliases (float seconds)
        "retrieval": search_s,
        "prompt_build": context_s,
        "cache_lookup": cache_lookup_s,
        "persist": cache_persist_s,
        # Renderer-shaped millisecond ints
        "search_ms": int(search_s * 1000),
        "context_ms": int(context_s * 1000),
        "llm_ms": int(llm_s * 1000),
        "verify_ms": int(verify_s * 1000),
        "cache_lookup_ms": int(cache_lookup_s * 1000),
        "persist_ms": int(cache_persist_s * 1000),
        "total_ms": int(total_s * 1000),
    }