ThothII/harness/tests/test_evidence_facade_contract.py

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),
    ]