"""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 json 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_persisted_merkle_proof_has_context_root_and_sources(setup): """merkle_proof column must hold the legacy-shape proof_obj: {context_root, sources: [{document_root, ...}], retrieval_purity}. arborist-viz Merkle Command Center reads this to synthesize the leaf set under a multi-source context root — an empty list shows 0 leaves in the 3D lattice.""" 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, ) qaro = sqlite3.connect(f"file:{qa}?mode=ro", uri=True) proof_str = qaro.execute( "SELECT merkle_proof FROM providence_cache WHERE cache_key = ?", (r["cache_key"],), ).fetchone()[0] qaro.close() proof = json.loads(proof_str) assert isinstance(proof, dict), "merkle_proof must be a dict (was list/empty)" assert proof["context_root"] == r["context_root"] assert len(proof["sources"]) >= 1 s0 = proof["sources"][0] assert s0["document_root"] assert "title" in s0 and "document_uri" in s0 assert "source_role" in s0 assert "used" in s0 and "used_pointer_ids" in s0 assert "retrieval_purity" in proof rp = proof["retrieval_purity"] assert rp["total_sources"] == len(proof["sources"]) 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]