diff --git a/arborist/qa/providence_query.py b/arborist/qa/providence_query.py index 8f72bdd..ba62961 100644 --- a/arborist/qa/providence_query.py +++ b/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( - question=question, - evidence_text=evidence_repr, - policy=policy, - model_id=model_id, + # 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 AFTER computing the key. Future revisions - # should reorder: compute a deterministic cache_key BEFORE - # retrieval (using source_root + question + policy only, plus - # 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. + # 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). 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 - # blob would need to be parsed + "sources": sources, + "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 diff --git a/tests/test_providence_query.py b/tests/test_providence_query.py new file mode 100644 index 0000000..38da9c4 --- /dev/null +++ b/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]