diff --git a/harness/tests/test_evidence_facade_contract.py b/harness/tests/test_evidence_facade_contract.py new file mode 100644 index 00000000..4564f2a8 --- /dev/null +++ b/harness/tests/test_evidence_facade_contract.py @@ -0,0 +1,199 @@ +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, + 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 _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 _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), + ] diff --git a/harness/tht/evidence/__init__.py b/harness/tht/evidence/__init__.py index e69de29b..af43dd6f 100644 --- a/harness/tht/evidence/__init__.py +++ b/harness/tht/evidence/__init__.py @@ -0,0 +1,97 @@ +"""Cohesive public entrypoint for Evidence domain capabilities. + +The implementation is introduced beside the legacy module layout so production callers can +migrate one path at a time. These functions deliberately preserve the existing objects, ordering, +and exceptions; they do not define a cross-domain service protocol. +""" + +from __future__ import annotations + +from collections.abc import Iterable +from pathlib import Path +from typing import TYPE_CHECKING + +from tht.ports.evidence import ( + AcquiredDocument, + EvidenceSource, + EvidenceSourceError, + EvidenceSourceErrorCategory, + SourceObject, +) + +if TYPE_CHECKING: + from tht.corpus.store import CorpusStore + from tht.decisions import DecisionRecord + from tht.search.evidence import ActiveEvidenceSearcher + from tht.session.models import SchemaLinking + + +def discover(source: EvidenceSource) -> Iterable[SourceObject]: + """Discover source objects without changing source-defined laziness or ordering.""" + return source.discover() + + +def acquire(source: EvidenceSource, item: SourceObject) -> AcquiredDocument: + """Acquire one discovered object, preserving the source's classified failures.""" + return source.acquire(item) + + +def active_searcher( + cfg, + delegate, + *, + workspace_id: str | None = None, +) -> ActiveEvidenceSearcher: + """Bind vector search to the atomically ACTIVE Evidence corpus generation.""" + from tht.search.evidence import active_searcher as legacy_active_searcher + + return legacy_active_searcher(cfg, delegate, workspace_id=workspace_id) + + +def validate_corpus_workspace(cfg, workspace_id: str) -> None: + """Validate persisted Evidence corpus ownership before runtime retrieval setup.""" + from tht.search.evidence import validate_corpus_workspace as legacy_validate + + legacy_validate(cfg, workspace_id) + + +def resolve_citation( + store: "CorpusStore", + evidence_id: str, + *, + materialized_root: Path | None = None, +) -> str: + """Resolve an Evidence identifier to its immutable ACTIVE materialization.""" + from tht.search.evidence import resolve_evidence_file + + return resolve_evidence_file( + store, + evidence_id, + materialized_root=materialized_root, + ) + + +def project_session( + decisions: list["DecisionRecord"], + linking: "SchemaLinking", + evidence_root: Path, +) -> list[dict]: + """Project cited and reviewed Evidence into the existing session artifact shape.""" + from tht.session.artifacts import build_evidence_entries + + return build_evidence_entries(decisions, linking, evidence_root) + + +__all__ = [ + "AcquiredDocument", + "EvidenceSource", + "EvidenceSourceError", + "EvidenceSourceErrorCategory", + "SourceObject", + "acquire", + "active_searcher", + "discover", + "project_session", + "resolve_citation", + "validate_corpus_workspace", +]