from datetime import UTC, datetime import hashlib import inspect from pathlib import Path from types import SimpleNamespace import pytest from tht.evidence.corpus.models import CanonicalDocument, CorpusManifest from tht.evidence.corpus.store import CorpusStore from tht.decisions import DecisionRecord from tht.evidence import ( acquire, active_searcher, build_preprocessing_pipeline, build_retrieval_entries, build_sources, discover, project_session, resolve_citation, ) from tht.evidence.contracts import ( AcquiredDocument, EvidenceSourceError, EvidenceSourceErrorCategory, SourceObject, ) 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 test_source_factory_preserves_legacy_first_order_and_filesystem_configuration(tmp_path): legacy_root = tmp_path / "legacy" configured_root = tmp_path / "configured" (legacy_root / "evidence").mkdir(parents=True) configured_root.mkdir() cfg = SimpleNamespace(evidence=SimpleNamespace( source_root=legacy_root, evidence_dir="evidence", sources=[SimpleNamespace( type="filesystem", root=configured_root, patterns=("*.md",), max_bytes=1024, )], )) current = build_sources(cfg.evidence) assert [source.root for source in current] == [ (legacy_root / "evidence").resolve(), configured_root.resolve(), ] assert current[1].patterns == ("*.md",) assert current[1].max_bytes == 1024 def test_preprocessing_factory_forwards_only_evidence_pipeline_dependencies(monkeypatch): captured = {} class FakePipeline: def __init__(self, **kwargs): captured.update(kwargs) monkeypatch.setattr("tht.evidence.preprocessing.CorpusPipeline", FakePipeline) dependencies = { "store": object(), "sources": [object()], "embedder": object(), "vector_store": object(), "embedding_model": "model", "embedding_dimensions": 3, "chunk_policy": object(), "pipeline_version": "evidence-v1", "retain_published_generations": 2, "workspace_id": None, } pipeline = build_preprocessing_pipeline(**dependencies) assert isinstance(pipeline, FakePipeline) assert captured == dependencies assert all( parameter.kind is not inspect.Parameter.VAR_KEYWORD for parameter in inspect.signature(build_preprocessing_pipeline).parameters.values() ) 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) facade_delegate = OrderedDelegate() 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] == ["higher", "lower"] assert facade_delegate.calls == [([1.0], 2, ["memory"], None)] 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" current = resolve_citation( store, "evi-used", materialized_root=materialized, ) assert current.endswith(".md") assert Path(current).read_text(encoding="utf-8") == "# evi-used\n" 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.session.resolve_citation", fake_resolve) linking = SchemaLinking( question="q", candidates=[Candidate( kind="table", name="fact_procedure", evidence=["evi-used"], decision="promoted", decision_seq=17, )], ) assert project_session([], 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, )], ) current = project_session(decisions, linking, evidence_root) assert [(row["id"], row["esito"], row["decision_seq"]) for row in current] == [ ("used", "usata", 17), ("accepted", "accettata", 21), ("rejected", "scartata", 22), ]