import hashlib import inspect from datetime import UTC, datetime from pathlib import Path from types import SimpleNamespace import pytest from tht.decisions import DecisionRecord from tht.evidence import ( EvidenceSearchContext, acquire, active_searcher, build_preprocessing_pipeline, build_retrieval_entries, build_sources, discover, project_session, resolve_citation, search_evidence, ) from tht.evidence.contracts import ( AcquiredDocument, EvidenceSourceError, EvidenceSourceErrorCategory, SourceObject, ) from tht.evidence.corpus.models import CanonicalDocument, CorpusManifest from tht.evidence.corpus.store import CorpusStore from tht.ports.vector import VectorReadUnavailable 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( local_archive_root=None, 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_legacy_source_discovers_only_curated_evidence_units(tmp_path): legacy_root = tmp_path / "legacy" (legacy_root / "evidence" / "source").mkdir(parents=True) (legacy_root / "evidence" / "source" / "raw.md").write_text("raw source") (legacy_root / "evidence" / "curated" / "domain").mkdir(parents=True) (legacy_root / "evidence" / "curated" / "domain" / "patient.md").write_text("curated") cfg = SimpleNamespace(evidence=SimpleNamespace( local_archive_root=None, source_root=legacy_root, evidence_dir="evidence", sources=[], )) source = build_sources(cfg.evidence)[0] assert [item.metadata["relative_path"] for item in source.discover()] == [ "curated/domain/patient.md" ] 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_id": "ollama/model", "embedding_model": "model", "embedding_dimensions": 3, "chunk_policy": object(), "pipeline_version": "evidence-v1", "retain_published_generations": 2, "workspace_id": None, "sparse_language": "italian", "candidate_evaluator": 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 test_typed_search_renders_one_stable_query_and_groups_fragments_by_evidence_unit(): """Removing context rendering, hard filters, or grouping changes this public result.""" class Searcher: vector_generation = "gen:" + "a" * 32 def __init__(self): self.calls = [] def search(self, embedding, **kwargs): self.calls.append((embedding, kwargs)) return [ SimpleNamespace( id="fragment:second", similarity=0.7, content="second excerpt", title="Pediatric range", metadata={ "evidence_id": "evidence:pediatric-range", "evidence_kind": "formula", "document_id": "doc:range", "ordinal": 1, "source_uri": "file:///curated/pediatric-range.md", "provenance": {"source_file": "source/range.md"}, }, ), SimpleNamespace( id="fragment:first", similarity=0.9, content="first excerpt", title="Pediatric range", metadata={ "evidence_id": "evidence:pediatric-range", "evidence_kind": "formula", "document_id": "doc:range", "ordinal": 0, "source_uri": "file:///curated/pediatric-range.md", "provenance": {"source_file": "source/range.md"}, }, ), ] class Embedder: def __init__(self): self.queries = [] def embed_query(self, query): self.queries.append(query) return [0.25] searcher = Searcher() embedder = Embedder() outcome = search_evidence( " Pazienti \"Età" + "\r\n" + " pediatrica ", "schema_linking", EvidenceSearchContext( concepts=("pediatrica", "pediatrica", " Età "), tables=("clinical.patient",), columns=("clinical.patient.Age",), required_kinds=("formula",), required_concepts=("Età",), required_tables=("clinical.patient",), required_columns=("clinical.patient.Age",), ), searcher=searcher, embedder=embedder, ) rendered = ( "Domanda: Pazienti \"Età\n pediatrica\n" "Concetti: Età, pediatrica\n" "Tabelle: clinical.patient\n" "Colonne: clinical.patient.Age" ) assert embedder.queries == [rendered] assert searcher.calls == [([0.25], { "top_n": 10, "kinds": ["evidence"], "query_text": rendered, "metadata_filter": { "purpose": "schema_linking", "required_kinds": ["formula"], "required_concepts": ["Età"], "required_tables": ["clinical.patient"], "required_columns": ["clinical.patient.Age"], }, })] assert outcome.status == "available" assert outcome.vector_generation == "gen:" + "a" * 32 assert [(item.evidence_id, item.excerpts, item.provenance, item.citation) for item in outcome.results] == [ ("evidence:pediatric-range", ("first excerpt", "second excerpt"), {"source_file": "source/range.md"}, "file:///curated/pediatric-range.md"), ] def test_typed_search_reports_vector_errors_as_unavailable_not_empty_results(): class UnavailableSearcher: vector_generation = "gen:" + "a" * 32 def search(self, _embedding, **_kwargs): raise VectorReadUnavailable("reader unavailable") outcome = search_evidence( "question", "rewriting", EvidenceSearchContext(), searcher=UnavailableSearcher(), embedder=SimpleNamespace(embed_query=lambda _query: [0.25]), ) assert outcome.status == "unavailable" assert outcome.code == "vector_unavailable" assert outcome.results == () def test_typed_search_without_an_active_generation_is_unavailable_not_an_empty_search(): outcome = search_evidence( "question", "rewriting", EvidenceSearchContext(), searcher=SimpleNamespace(), embedder=SimpleNamespace(embed_query=lambda _query: [0.25]), ) assert outcome.status == "unavailable" assert outcome.code == "active_corpus_unavailable" def test_typed_search_reports_a_malformed_fragment_payload_as_unavailable(): class Searcher: vector_generation = "gen:" + "a" * 32 def search(self, _embedding, **_kwargs): return [SimpleNamespace( id="fragment:bad", similarity=0.5, title="Bad", content="bad", metadata={"evidence_id": "evidence:bad"}, )] outcome = search_evidence( "question", "rewriting", EvidenceSearchContext(), searcher=Searcher(), embedder=SimpleNamespace(embed_query=lambda _query: [0.25]), ) assert outcome.status == "unavailable" assert outcome.code == "evidence_search_unavailable" 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), ]