arborist/tests/test_providence_source.py

"""Tests for ProvidenceSource — self-reference thought chains.

Covers the four iteration-time gates documented in
docs/self-reference-thought-chains-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 `aborist://providence/`
URIs classify as `self_reference_source`.
"""

from __future__ import annotations

import sqlite3
import time

import pytest

from aborist.qa.query import _classify_source_role
from aborist.sources.providence import (
    PROVIDENCE_URI_PREFIX,
    ProvidenceSource,
)
from aborist.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
    `aborist://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")