"""Tests for ProvidenceSource — self-reference thought chains. Covers the four iteration-time gates documented in docs/self-reference-design.md: 1. audit_mode == 'STRICT' (HYBRID/UNGROUNDED excluded) 2. falsification_state == 'live' (failed/stale/quarantined excluded) 3. now - created_at >= kindergarten_seconds (fresh records cool first) 4. anti-recursion: records whose answer cites a self-reference URI are excluded (first-generation only) Plus the classifier-side: documents with `arborist://providence/` URIs classify as `self_reference_source`. """ from __future__ import annotations import sqlite3 import time import pytest from arborist.qa.query import _classify_source_role from arborist.sources.providence import ( PROVIDENCE_URI_PREFIX, ProvidenceSource, ) from arborist.store import connect def _seed(conn: sqlite3.Connection, **fields) -> str: """Insert one minimal providence_cache row. Returns the cache_key.""" defaults = { "cache_key": fields.get("cache_key", "ck_" + str(int(time.time() * 1e6))), "source_root": "00" * 32, "document_uri": "test://doc", "question_hash": "11" * 32, "question_text": "stub question", "answer_text": "stub answer", "merkle_proof": '{"proofs": []}', "model_profile_hash": "22" * 32, "conversation_hash": "33" * 32, "governance_policy_hash": "44" * 32, "schema_version": "v9.8.0", "canonicalization_version": "norm-v1", "chunking_version": "tok-512-v1", "falsification_state": "live", "chain": "private", "audit_event_hash": "55" * 32, "created_at": time.time() - 7200, # 2h old by default — past kindergarten "last_hit_at": None, "hit_count": 0, "audit_mode": "STRICT", "n_quotes": 1, "n_verified": 1, "unverified_quotes": None, "verifier_method": "claim_lattice", "run_dag_root": "66" * 32, "run_dag_blob": "{}", } defaults.update(fields) cols = ", ".join(defaults.keys()) placeholders = ", ".join("?" * len(defaults)) conn.execute( f"INSERT INTO providence_cache ({cols}) VALUES ({placeholders})", tuple(defaults.values()), ) conn.commit() return defaults["cache_key"] def test_yields_strict_live_record_past_kindergarten(tmp_path): db = tmp_path / "p.db" conn = connect(db) try: ck = _seed(conn, cache_key="strict-cooled", question_text="who is X?", answer_text="X is Y.") # Source created with kindergarten=3600. Record is 2h old → eligible. src = ProvidenceSource(conn, kindergarten_seconds=3600) docs = list(src.iter_documents()) assert len(docs) == 1 d = docs[0] assert d.uri == f"{PROVIDENCE_URI_PREFIX}{ck}" assert d.source_type == "providence" assert d.title == "who is X?" assert "Q: who is X?" in d.content assert "A: X is Y." in d.content finally: conn.close() def test_excludes_hybrid_and_ungrounded(tmp_path): db = tmp_path / "p.db" conn = connect(db) try: _seed(conn, cache_key="strict-1", audit_mode="STRICT") _seed(conn, cache_key="hybrid-1", audit_mode="HYBRID") _seed(conn, cache_key="ungrounded-1", audit_mode="UNGROUNDED") docs = list(ProvidenceSource(conn).iter_documents()) uris = {d.uri for d in docs} assert f"{PROVIDENCE_URI_PREFIX}strict-1" in uris assert f"{PROVIDENCE_URI_PREFIX}hybrid-1" not in uris assert f"{PROVIDENCE_URI_PREFIX}ungrounded-1" not in uris finally: conn.close() def test_excludes_falsified_records(tmp_path): db = tmp_path / "p.db" conn = connect(db) try: _seed(conn, cache_key="live-strict", audit_mode="STRICT", falsification_state="live") _seed(conn, cache_key="failed-strict", audit_mode="STRICT", falsification_state="failed") _seed(conn, cache_key="stale-strict", audit_mode="STRICT", falsification_state="stale") _seed(conn, cache_key="quarantined-strict", audit_mode="STRICT", falsification_state="quarantined") docs = list(ProvidenceSource(conn).iter_documents()) uris = {d.uri for d in docs} assert f"{PROVIDENCE_URI_PREFIX}live-strict" in uris # Falsified rows excluded — verifier-falsification mechanism # works exactly because state=live is the gate. assert f"{PROVIDENCE_URI_PREFIX}failed-strict" not in uris assert f"{PROVIDENCE_URI_PREFIX}stale-strict" not in uris assert f"{PROVIDENCE_URI_PREFIX}quarantined-strict" not in uris finally: conn.close() def test_kindergarten_window_excludes_fresh_records(tmp_path): db = tmp_path / "p.db" conn = connect(db) try: now = time.time() # 30s old — fresh, inside the 1h window. _seed(conn, cache_key="fresh", created_at=now - 30) # 2h old — past the window. _seed(conn, cache_key="cooled", created_at=now - 7200) src = ProvidenceSource(conn, kindergarten_seconds=3600, now_seconds=now) docs = list(src.iter_documents()) uris = {d.uri for d in docs} assert f"{PROVIDENCE_URI_PREFIX}cooled" in uris assert f"{PROVIDENCE_URI_PREFIX}fresh" not in uris finally: conn.close() def test_kindergarten_zero_admits_all_strict_live(tmp_path): db = tmp_path / "p.db" conn = connect(db) try: now = time.time() _seed(conn, cache_key="just-now", created_at=now - 1) _seed(conn, cache_key="cooled", created_at=now - 86400) src = ProvidenceSource(conn, kindergarten_seconds=0, now_seconds=now) docs = list(src.iter_documents()) assert len(docs) == 2 finally: conn.close() def test_anti_recursion_excludes_self_referencing_records(tmp_path): """A STRICT record whose own answer text contains `arborist://providence/...` is excluded — first-generation only. Prevents echo-chamber chains where a wrong-but-STRICT record keeps getting recompiled into deeper claims.""" db = tmp_path / "p.db" conn = connect(db) try: _seed(conn, cache_key="first-gen", answer_text="X is Y per primary source.") _seed(conn, cache_key="second-gen", answer_text=f"X is Y per {PROVIDENCE_URI_PREFIX}other-key cited record.") docs = list(ProvidenceSource(conn).iter_documents()) uris = {d.uri for d in docs} assert f"{PROVIDENCE_URI_PREFIX}first-gen" in uris assert f"{PROVIDENCE_URI_PREFIX}second-gen" not in uris finally: conn.close() def test_skips_records_with_empty_question_or_answer(tmp_path): db = tmp_path / "p.db" conn = connect(db) try: _seed(conn, cache_key="no-q", question_text="", answer_text="A") _seed(conn, cache_key="no-a", question_text="Q", answer_text="") _seed(conn, cache_key="both", question_text="Q", answer_text="A") docs = list(ProvidenceSource(conn).iter_documents()) uris = {d.uri for d in docs} assert f"{PROVIDENCE_URI_PREFIX}both" in uris assert f"{PROVIDENCE_URI_PREFIX}no-q" not in uris assert f"{PROVIDENCE_URI_PREFIX}no-a" not in uris finally: conn.close() def test_classify_source_role_recognizes_providence_uri(): """`_classify_source_role` short-circuits on the URI scheme regardless of title shape. Trust model is URI-based, not title-heuristic-based.""" qstem = {"foo", "bar"} role = _classify_source_role( "Some Title (film)", qstem, document_uri=f"{PROVIDENCE_URI_PREFIX}abc123", ) assert role == "self_reference_source" def test_classify_source_role_falls_through_for_non_providence_uri(): """External URIs get the existing title-based classification.""" qstem = {"jurassic", "park", "film"} role = _classify_source_role( "Jurassic Park (film)", qstem, document_uri="https://en.wikipedia.org/wiki/Jurassic_Park_(film)", ) # Title has 3 stems matching the 3-stem query → primary. assert role == "primary_answer_source" def test_classify_source_role_handles_missing_uri(): """document_uri is optional; without it the function falls back to the existing title-based classification (backward compat).""" qstem = {"jurassic", "park"} role = _classify_source_role("Jurassic Park (film)", qstem) # Stems match, primary classification. assert role in ("primary_answer_source", "background_source")