qa/providence_query: cache persist on miss + audit chain append

Phase 2 step 2 of #000072. The skeleton landed in 20faae0 looked up
providence_cache + returned on hit, but the miss branch just
returned the fresh run_query result without writing anything back —
so every call paid the LLM cost. Now misses persist:

  1. Build run_dag via arborist.qa.dag.build_run_dag (claim-lattice
     mode, 9-stage variant — same shape legacy query() emits)
  2. Append a providence_write event via arborist.store.append_audit
     INSIDE a BEGIN IMMEDIATE transaction
  3. INSERT INTO providence_cache with all 25 columns the schema
     requires (cache_key, source_root, document_uri, question_*,
     answer_text, merkle_proof placeholder, 4-dim policy hashes,
     3-dim schema versions, audit_event_hash linking to the just-
     appended event, run_dag_root + run_dag_blob, audit_mode, etc.)
  4. COMMIT — atomic; audit chain + providence_cache stay
     consistent on crash mid-way

Result dict on miss now carries audit_event_hash + run_dag_root so
the caller can reference the audit chain or replay the DAG.
status="fresh_persisted" (was "fresh" in the skeleton) names the
new behavior.

NOT done yet (deferred to subsequent steps):
  - merkle_proof is "[]" placeholder. Schema requires NOT NULL.
    Real per-chunk proofs are a follow-up; the cache row is
    consistent without them but downstream wallet verification has
    nothing to walk.
  - burn_existing doesn't emit a providence_burn audit event
    (legacy query() does — query.py:3084 region). Audit chain
    still grows monotonically on the persist side, just doesn't
    record what was burned.
  - equivalence_class fallback lookup (legacy tries both dedup-mode
    keys when fidelity allows; primary only here)

Tests (tests/test_providence_query.py, 4):
  - first call persists row + audit event
  - second call returns cache_hit ignoring different stub
  - burn_existing forces re-run + re-persist; audit chain grows
  - chain links correctly across two distinct cache_keys

283 tests pass in the broader query/corpus/sidecar/wallet/bucket/
claim_lattice/byte_identity/providence gate.

arborist/qa/providence_query.py

@@ -32,9 +32,12 @@ 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,
@@ -46,6 +49,7 @@ 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
@@ -155,6 +159,115 @@ def _burn(qa_conn: _sqlite.Connection, ckey: str) -> int:
     return cur.rowcount
 
 
+def _persist_miss(
+    qa_conn: _sqlite.Connection,
+    *,
+    ckey: str,
+    question: str,
+    run_result: dict,
+    sources: list,
+    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).
+
+    Minimal-viable persist: merkle_proof is an empty JSON array
+    placeholder. Real per-chunk Merkle proofs are a follow-up — the
+    schema requires the column NOT NULL, so empty array satisfies
+    that without claiming proof structure the caller didn't actually
+    build.
+
+    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, "[]",
+                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,
@@ -218,21 +331,38 @@ def providence_query(
         for s in sources
     )
     source_root = corpus.snapshot_root()
-    ckey = _compute_cache_key(
+    # Pre-compute each dimension once — needed both for cache_key
-        question=question,
+    # AND for the persist path below.
-        evidence_text=evidence_repr,
+    qhash = question_hash(question, mode="equivalence_class")
-        policy=policy,
+    mhash = model_profile_hash(model_id)
-        model_id=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 AFTER computing the key. Future revisions
+    # 4. Cache lookup. Burn first if requested (so a re-lookup after
-    # should reorder: compute a deterministic cache_key BEFORE
+    # burn correctly misses and re-runs the LLM in the SAME call —
-    # retrieval (using source_root + question + policy only, plus
+    # legacy convention).
-    # a separate context_hash dimension) so burn + hit can happen
-    # before the LLM call. Today we accept the LLM-after-lookup
-    # cost as the trade-off for the minimal skeleton.
     qa_conn = _qa_conn(qa_db)
+    audit_event_hash = None
+    run_dag_root = None
     try:
         if burn_existing:
             burned_existing = _burn(qa_conn, ckey)
@@ -245,26 +375,36 @@ def providence_query(
                 "lookup_path": "primary",
                 "audit_mode": cached["audit_mode"],
                 "answer_text": cached["answer_text"],
-                "sources": sources,  # from this run; stored sources
+                "sources": sources,
-                                     # blob would need to be parsed
+                "audit_event_hash": cached["audit_event_hash"],
+                "run_dag_root": cached["run_dag_root"],
                 "burned_existing": 0,
                 "elapsed_s": round(_time.time() - t_total, 3),
             }
+        # 5. Miss — persist the fresh result + append audit. ckey
+        # is the index; the row holds run_dag_blob + audit_event_hash
+        # so a future re-lookup can validate the chain.
+        audit_event_hash, run_dag = _persist_miss(
+            qa_conn,
+            ckey=ckey, question=question, run_result=run_result,
+            sources=sources, 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"]
     finally:
         qa_conn.close()
 
-    # 5. Fresh result — return with cache metadata. Persist is
-    # deferred to legacy query()'s flow for now (the schema requires
-    # many fields beyond what run_query produces: run_dag_root,
-    # context_root, prompt_hash, verifier_method, n_quotes,
-    # n_verified, violations_json, etc.). Future revisions add a
-    # full persist path here so providence_query can stand alone.
     out = dict(run_result)
     out.update({
-        "status": "burned" if burned_existing else "fresh",
+        "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),
     })
     return out

tests/test_providence_query.py

@@ -0,0 +1,159 @@
+"""End-to-end tests for arborist.qa.providence_query.
+
+Covers Phase 2 step 1 + step 2 of #000072:
+  - Step 1: cache hit / miss / burn shape (skeleton landed 20faae0)
+  - Step 2: persist on miss — providence_cache row + audit chain
+    event written atomically inside one transaction
+"""
+from __future__ import annotations
+
+import sqlite3
+from pathlib import Path
+from typing import Iterator
+
+import pytest
+
+from arborist.document import Document
+from arborist.ingest import ingest_source
+from arborist.qa.client import StubClient
+from arborist.qa.corpus import SqliteShardCorpus
+from arborist.qa.providence_query import providence_query
+from arborist.source import Source
+from arborist.store import connect
+
+
+class _S(Source):
+    source_type = "test"
+    def __init__(self, ds): self.ds = ds
+    def iter_documents(self) -> Iterator[Document]:
+        yield from self.ds
+
+
+@pytest.fixture
+def setup(tmp_path):
+    """Build a one-doc corpus + an empty qa.db; return (corpus, qa_path)."""
+    shard = tmp_path / "shard.db"
+    qa = tmp_path / "qa.db"
+    c = connect(shard)
+    ingest_source(c, _S([Document(
+        uri="test://anarchism", source_type="test", title="Anarchism",
+        content=(
+            "Anarchism is a political philosophy that promotes a stateless "
+            "society. " * 6
+            + 'The phrase "abolition of authority" describes the core principle. ' * 4
+        ),
+    )]))
+    c.execute("PRAGMA wal_checkpoint(FULL)")
+    c.close()
+    connect(qa).close()  # init schema
+    corpus = SqliteShardCorpus(connect(shard))
+    return corpus, qa
+
+
+def test_first_call_persists_with_audit_event(setup):
+    corpus, qa = setup
+    stub = 'The core principle of anarchism is the "abolition of authority". [E1]'
+    r = providence_query(
+        corpus, "what is the core principle of anarchism?",
+        StubClient(answer=stub),
+        qa_db=qa, model_id="stub", top_k=2,
+    )
+    assert r["status"] == "fresh_persisted"
+    assert r["audit_mode"] == "STRICT"
+    assert r["cache_key"]
+    assert r["audit_event_hash"]
+    assert r["run_dag_root"]
+    qaro = sqlite3.connect(f"file:{qa}?mode=ro", uri=True)
+    n_cache = qaro.execute(
+        "SELECT COUNT(*) FROM providence_cache WHERE cache_key = ?",
+        (r["cache_key"],),
+    ).fetchone()[0]
+    assert n_cache == 1
+    n_audit = qaro.execute(
+        "SELECT COUNT(*) FROM audit_events WHERE event_type='providence_write'"
+    ).fetchone()[0]
+    assert n_audit == 1
+    qaro.close()
+
+
+def test_second_call_returns_cache_hit(setup):
+    """Second call with the SAME question + model + corpus must hit
+    the cache and ignore the (different) stub answer."""
+    corpus, qa = setup
+    q = "what is the core principle of anarchism?"
+    stub1 = 'The core principle of anarchism is the "abolition of authority". [E1]'
+    r1 = providence_query(
+        corpus, q, StubClient(answer=stub1),
+        qa_db=qa, model_id="stub", top_k=2,
+    )
+    r2 = providence_query(
+        corpus, q, StubClient(answer="WRONG ANSWER [E1]"),
+        qa_db=qa, model_id="stub", top_k=2,
+    )
+    assert r2["status"] == "cache_hit"
+    assert r2["audit_mode"] == r1["audit_mode"]
+    assert r2["answer_text"] == r1["answer_text"]  # NOT the wrong stub
+    assert r2["cache_key"] == r1["cache_key"]
+
+
+def test_burn_existing_forces_re_run_and_re_persists(setup):
+    """burn_existing deletes the cached row, re-runs the LLM, and
+    persists a NEW row (which by chance may have the same cache_key
+    if inputs are identical — that's expected). audit chain grows by
+    one event per write."""
+    corpus, qa = setup
+    q = "what is the core principle of anarchism?"
+    stub = 'The core principle of anarchism is the "abolition of authority". [E1]'
+    providence_query(
+        corpus, q, StubClient(answer=stub),
+        qa_db=qa, model_id="stub", top_k=2,
+    )
+    r2 = providence_query(
+        corpus, q, StubClient(answer=stub),
+        qa_db=qa, model_id="stub", top_k=2, burn_existing=True,
+    )
+    assert r2["status"] == "burned"
+    assert r2["burned_existing"] == 1
+    # After burn + re-persist, exactly one live row + 2 audit events
+    # (write + write — the burn itself doesn't currently emit a
+    # 'providence_burn' event in providence_query; legacy query()
+    # does, but that's deferred to a later sub-step).
+    qaro = sqlite3.connect(f"file:{qa}?mode=ro", uri=True)
+    n_live = qaro.execute(
+        "SELECT COUNT(*) FROM providence_cache "
+        "WHERE falsification_state = 'live'"
+    ).fetchone()[0]
+    assert n_live == 1
+    n_audit = qaro.execute(
+        "SELECT COUNT(*) FROM audit_events WHERE event_type='providence_write'"
+    ).fetchone()[0]
+    assert n_audit == 2
+    qaro.close()
+
+
+def test_audit_chain_links_correctly(setup):
+    """Each providence_write event must chain off the previous
+    head — append_audit's own discipline. Verified by walking the
+    chain backwards from the latest event."""
+    corpus, qa = setup
+    q = "what is the core principle of anarchism?"
+    stub = 'The core principle of anarchism is the "abolition of authority". [E1]'
+    # Write two rows via two distinct cache_keys (different model_ids).
+    r1 = providence_query(
+        corpus, q, StubClient(answer=stub),
+        qa_db=qa, model_id="stub", top_k=2,
+    )
+    r2 = providence_query(
+        corpus, q, StubClient(answer=stub),
+        qa_db=qa, model_id="stub2", top_k=2,
+    )
+    assert r1["cache_key"] != r2["cache_key"]
+    qaro = sqlite3.connect(f"file:{qa}?mode=ro", uri=True)
+    rows = qaro.execute(
+        "SELECT event_hash, prev_event_hash, event_type FROM audit_events "
+        "WHERE event_type='providence_write' ORDER BY ts ASC"
+    ).fetchall()
+    qaro.close()
+    assert len(rows) == 2
+    # Second event's prev_event_hash must be the first event's hash.
+    assert rows[1][1] == rows[0][0]