from datetime import UTC, datetime import hashlib from types import SimpleNamespace import pytest from tht.corpus.models import CanonicalDocument, CorpusManifest from tht.corpus.store import CorpusStore from tht.decisions import DecisionRecord from tht.evidence import ( acquire, active_searcher, build_retrieval_entries, discover, project_session, resolve_citation, ) from tht.ports.evidence import ( AcquiredDocument, EvidenceSourceError, EvidenceSourceErrorCategory, SourceObject, ) from tht.search.evidence import active_searcher as legacy_active_searcher from tht.search.evidence import resolve_evidence_file from tht.session.artifacts import build_evidence_entries from tht.session.models import Candidate, SchemaLinking class RecordingSource: def __init__(self, *, fail=False): self.items = [ SourceObject( source_id="source:z", uri="https://example.test/z.md", fingerprint="sha256:z", ), SourceObject( source_id="source:a", uri="https://example.test/a.md", fingerprint="sha256:a", ), ] self.fail = fail self.calls = [] def discover(self): self.calls.append(("discover",)) return iter(self.items) def acquire(self, item): self.calls.append(("acquire", item.source_id)) if self.fail: raise EvidenceSourceError( "transport detail must stay hidden", category=EvidenceSourceErrorCategory.TRANSIENT, details={"operation": "download"}, ) return AcquiredDocument(source=item, content=item.source_id.encode()) def test_acquisition_facade_preserves_source_order_results_and_calls(): legacy = RecordingSource() facade = RecordingSource() legacy_items = list(legacy.discover()) facade_items = list(discover(facade)) assert facade_items == legacy_items assert [item.source_id for item in facade_items] == ["source:z", "source:a"] assert acquire(facade, facade_items[0]) == legacy.acquire(legacy_items[0]) assert facade.calls == legacy.calls == [ ("discover",), ("acquire", "source:z"), ] def test_acquisition_facade_preserves_classified_errors(): source = RecordingSource(fail=True) with pytest.raises(EvidenceSourceError) as captured: acquire(source, source.items[0]) assert str(captured.value) == "evidence source operation failed" assert captured.value.category is EvidenceSourceErrorCategory.TRANSIENT assert captured.value.retryable is True assert captured.value.details == {"operation": "download"} def _active_config(tmp_path): store = CorpusStore(tmp_path / "corpus") generation = store.stage( CorpusManifest(metadata={"workspace_id": "workspace-a"}), {}, generation="gen:" + "a" * 32, ) store.publish(generation) return SimpleNamespace(paths=SimpleNamespace(artifacts=tmp_path / "artifacts")) class OrderedDelegate: def __init__(self): self.calls = [] def search(self, embedding, top_n=10, kinds=None, metadata_filter=None): self.calls.append((embedding, top_n, kinds, metadata_filter)) return [ SimpleNamespace(id="lower", similarity=0.4), SimpleNamespace(id="higher", similarity=0.9), ] def test_search_facade_preserves_active_filtering_and_global_order(tmp_path): cfg = _active_config(tmp_path) legacy_delegate = OrderedDelegate() facade_delegate = OrderedDelegate() legacy = legacy_active_searcher( cfg, legacy_delegate, workspace_id="workspace-a", ).search([1.0], top_n=2, kinds=["evidence", "memory"]) current = active_searcher( cfg, facade_delegate, workspace_id="workspace-a", ).search([1.0], top_n=2, kinds=["evidence", "memory"]) assert [hit.id for hit in current] == [hit.id for hit in legacy] == ["higher", "lower"] assert facade_delegate.calls == legacy_delegate.calls def test_retrieval_entries_preserve_hit_order_and_existing_projection_shape(): hits = [ SimpleNamespace(label="Second", status="reviewed", content="abcdefgh"), SimpleNamespace(label="First", status=None, content="12345678"), ] assert build_retrieval_entries(hits, excerpt_chars=5) == [ {"title": "Second", "status": "reviewed", "excerpt": "abcde"}, {"title": "First", "status": None, "excerpt": "12345"}, ] def _canonical_store(root, evidence_id): content = f"# {evidence_id}\n" digest = hashlib.sha256(content.encode()).hexdigest() document = CanonicalDocument( document_id=f"doc:{digest}", source_id=f"source:{evidence_id}", source_uri=f"file:///curated/{evidence_id}.md", source_fingerprint="sha256:" + "b" * 64, content_hash=f"sha256:{digest}", content=content, pipeline_version="evidence-v1", metadata={"frontmatter": {"id": evidence_id}}, ) store = CorpusStore(root) generation = store.stage(CorpusManifest(documents=(document,)), {document.document_id: content}) store.publish(generation) return store def test_citation_facade_matches_active_corpus_resolution(tmp_path): store = _canonical_store(tmp_path / "corpus", "evi-used") materialized = tmp_path / "materialized" legacy = resolve_evidence_file( store, "evi-used", materialized_root=materialized, ) current = resolve_citation( store, "evi-used", materialized_root=materialized, ) assert current == legacy assert resolve_citation(store, "missing", materialized_root=materialized) == "" def test_session_projection_routes_corpus_citations_through_the_facade(tmp_path, monkeypatch): evidence_root = tmp_path / "artifacts" / "evidence" (tmp_path / "corpus").mkdir() calls = [] def fake_resolve(store, evidence_id, *, materialized_root=None): calls.append((store.root, evidence_id, materialized_root)) return f"/materialized/{evidence_id}.md" monkeypatch.setattr("tht.evidence.resolve_citation", fake_resolve) linking = SchemaLinking( question="q", candidates=[Candidate( kind="table", name="fact_procedure", evidence=["evi-used"], decision="promoted", decision_seq=17, )], ) assert build_evidence_entries([], linking, evidence_root) == [{ "id": "evi-used", "file": "/materialized/evi-used.md", "esito": "usata", "decision_seq": 17, }] assert calls == [( tmp_path / "corpus", "evi-used", tmp_path / "artifacts" / ".materialized-evidence", )] def _record(seq, type_, subject): return DecisionRecord( seq=seq, ts=datetime(2026, 8, 24, tzinfo=UTC), type=type_, subject=subject, ) def test_session_projection_facade_preserves_outcome_precedence_and_order(tmp_path): evidence_root = tmp_path / "artifacts" / "evidence" evidence_root.mkdir(parents=True) for evidence_id in ("used", "accepted", "rejected"): (evidence_root / f"{evidence_id}.md").write_text(f"# {evidence_id}\n") decisions = [ _record(21, "evidence_accepted", "accepted"), _record(22, "evidence_rejected", "rejected"), ] linking = SchemaLinking( question="q", candidates=[Candidate( kind="table", name="fact_procedure", evidence=["used", "accepted"], decision="promoted", decision_seq=17, )], ) legacy = build_evidence_entries(decisions, linking, evidence_root) current = project_session(decisions, linking, evidence_root) assert current == legacy assert [(row["id"], row["esito"], row["decision_seq"]) for row in current] == [ ("used", "usata", 17), ("accepted", "accettata", 21), ("rejected", "scartata", 22), ]