ThothII/harness/tests/test_evidence_candidate_publication.py

"""The configured candidate evaluator must gate publication on its exact generation."""

import hashlib
from pathlib import Path
from types import SimpleNamespace

import pytest
from pydantic import ValidationError

from tht.cli import preprocess_cmd
from tht.config import VectorConfig, load_config
from tht.evidence import CuratedEvidence, EvidenceManifest, dump_curated_markdown, dump_manifest
from tht.evidence.authoring import ManifestSource
from tht.evidence.contracts import AcquiredDocument, SourceObject
from tht.evidence.corpus.chunk import ChunkPolicy
from tht.evidence.corpus.pipeline import CorpusPipeline, PipelineError
from tht.evidence.corpus.store import CorpusStore
from tht.ports.vector import VectorCapabilities, VectorHealth


class Source:
    def __init__(self, item: SourceObject, content: str) -> None:
        self.item = item
        self.content = content

    def discover(self):
        return [self.item]

    def acquire(self, item):
        assert item == self.item
        return AcquiredDocument(source=item, content=self.content.encode(), media_type="text/markdown")


class Embedder:
    def embed_documents(self, texts):
        return [[0.25] * 1024 for _ in texts]

    def embed_query(self, query):
        assert query
        return [0.25] * 1024


class ObservableStore(CorpusStore):
    def __init__(self, root: Path) -> None:
        super().__init__(root)
        self.events: list[tuple[str, str]] = []

    def publish(self, generation: str) -> str:
        self.events.append(("publish", generation))
        return super().publish(generation)


class ObservableVectors:
    capabilities = VectorCapabilities(search=True, existing_hashes=True, upsert=True)

    def __init__(self, store: ObservableStore) -> None:
        self.store = store
        self.records = []
        self.searches: list[dict] = []
        self.fail_evaluation = False

    def health(self):
        return VectorHealth(
            ok=True,
            expected_dimension=1024,
            observed_dimensions=(1024,),
            dimension_compatible=True,
        )

    def existing_hashes(self, _collection, _kinds):
        return {entry.record.id: entry.content_hash for entry in self.records}

    def upsert(self, _collection, records):
        self.records.extend(records)
        return len(records)

    def delete_generation(self, _collection, generation, _workspace_id):
        self.records = [
            entry for entry in self.records
            if entry.record.metadata["vector_generation"] != generation
        ]
        return 0

    def list_evidence_generations(self, _collection, _workspace_id):
        return sorted({entry.record.metadata["vector_generation"] for entry in self.records})

    def search(self, _collections, _embedding, **kwargs):
        metadata_filter = kwargs["metadata_filter"]
        generation = metadata_filter["vector_generation"]
        self.searches.append({
            "generation": generation,
            "mode": kwargs["retrieval_mode"],
            "active": self.store.active_generation(),
            "query": kwargs["query_text"],
        })
        assert self.store.active_generation() != generation
        if self.fail_evaluation:
            return []
        for entry in self.records:
            metadata = entry.record.metadata
            if metadata["vector_generation"] == generation:
                return [SimpleNamespace(
                    id=entry.record.id,
                    similarity=1.0,
                    metadata={
                        "evidence_id": metadata["evidence_id"],
                        "evidence_kind": metadata["evidence_kind"],
                    },
                )]
        return []


def _workspace(tmp_path: Path):
    workspace = tmp_path / "workspace"
    source_text = "Pazienti con età inferiore a 18 anni.\n"
    source_file = workspace / "evidence" / "source" / "notes.md"
    curated_file = workspace / "evidence" / "curated" / "formula" / "fascia-pediatrica.md"
    source_file.parent.mkdir(parents=True)
    curated_file.parent.mkdir(parents=True)
    source_file.write_text(source_text, encoding="utf-8")
    digest = "sha256:" + hashlib.sha256(source_text.encode()).hexdigest()
    evidence = CuratedEvidence.model_validate({
        "schema_version": 1,
        "id": "evidence:fascia-pediatrica",
        "title": "Fascia pediatrica",
        "kind": "formula",
        "purposes": ["sql_generation"],
        "applies_to": {"columns": ["clinical.patient.birth_date"]},
        "language": "it",
        "provenance": {
            "source_file": "source/notes.md",
            "source_sha256": digest,
            "supporting_excerpts": [source_text.strip()],
        },
        "review_items": [],
        "payload": {
            "concept": "fascia pediatrica",
            "columns": ["clinical.patient.birth_date"],
            "sql": "CASE WHEN age < 18 THEN 'pediatrica' ELSE 'adulta' END",
        },
    })
    curated = dump_curated_markdown(evidence)
    curated_file.write_text(curated, encoding="utf-8")
    (workspace / "evidence" / "manifest.yaml").write_text(dump_manifest(EvidenceManifest(
        schema_version=1,
        pipeline_version="evidence-authoring-v1",
        sources={"source/notes.md": ManifestSource(sha256=digest, units=(evidence.id,))},
        orphans=(),
    )), encoding="utf-8")
    (workspace / "evidence" / "evaluation.yaml").write_text(
        """
schema_version: 1
queries:
  - id: lexical
    query: CASE patient.birth_date
    profile: lexical
    purpose: sql_generation
    expected: [evidence:fascia-pediatrica]
  - id: semantic
    query: Quali pazienti sono pediatrici?
    profile: semantic
    purpose: sql_generation
    expected: [evidence:fascia-pediatrica]
  - id: mixed
    query: Formula per patient.birth_date pediatrica
    profile: mixed
    purpose: sql_generation
    expected: [evidence:fascia-pediatrica]
""".strip(),
        encoding="utf-8",
    )
    config = tmp_path / "workspace.yaml"
    config.write_text(
        f"""
runtime_identity:
  workspace_id: psd-clinical
  workspace_revision: {'a' * 40}
dwh:
  type: postgres_direct
  connection: {{database: analytics, schema: mart, user: reader, password: test-only}}
vectors:
  type: qdrant
  base_url: http://qdrant:6333
  collection: psd-clinical
embeddings:
  provider: ollama_internal
  base_url: http://embedding:11434
  model: qwen3-embedding:0.6b
  dim: 1024
evidence:
  schema_version: 2
  sources:
    - type: filesystem
      root: {workspace / 'evidence'}
roots:
  sessions: {tmp_path / 'sessions'}
  artifacts: {tmp_path / 'artifacts'}
  indexes: {tmp_path / 'indexes'}
""".strip(),
        encoding="utf-8",
    )
    item = SourceObject(
        source_id="fs:curated-formula",
        uri=curated_file.as_uri(),
        fingerprint="sha256:" + "b" * 64,
        metadata={"relative_path": "curated/formula/fascia-pediatrica.md"},
    )
    return load_config(config), Source(item, curated), item


def _pipeline(store, source, vectors, evaluator):
    return CorpusPipeline(
        store=store,
        sources=[source],
        embedder=Embedder(),
        vector_store=vectors,
        embedding_model="test-model",
        embedding_dimensions=1024,
        chunk_policy=ChunkPolicy(version="semantic:v1", max_chars=4000),
        pipeline_version="evidence-v1",
        workspace_id="psd-clinical",
        candidate_evaluator=evaluator,
    )


def test_validated_corpus_evaluates_exact_inactive_generation_before_publication(tmp_path):
    cfg, source, item = _workspace(tmp_path)
    preprocess_cmd._validate_materialized_curated_corpus(cfg)
    store = ObservableStore(tmp_path / "corpus")
    vectors = ObservableVectors(store)
    evaluator = preprocess_cmd._candidate_evaluator(cfg, vector_store=vectors, embedder=Embedder())

    first = _pipeline(store, source, vectors, evaluator).run()

    assert first.published is True
    assert store.active_generation() == first.generation
    assert len(vectors.searches) == 9
    assert {call["mode"] for call in vectors.searches} == {"dense", "bm25", "fused"}
    assert {call["generation"] for call in vectors.searches} == {first.generation}
    assert {call["active"] for call in vectors.searches} == {None}
    assert store.events == [("publish", first.generation)]
    assert VectorConfig().max_chunk_chars == 4000
    assert set(VectorConfig.model_fields) == {"max_chunk_chars", "retain_published_generations"}
    for alias in ("chunk_size", "max_chunk_size", "max_fragment_chars"):
        with pytest.raises(ValidationError, match="extra_forbidden"):
            VectorConfig.model_validate({alias: 4000})
    assert all(len(fragment.content) <= VectorConfig().max_chunk_chars for fragment in first.manifest.chunks)

    vectors.fail_evaluation = True
    changed = SourceObject(
        source_id=item.source_id,
        uri=item.uri,
        fingerprint="sha256:" + "c" * 64,
        metadata=item.metadata,
    )
    with pytest.raises(PipelineError, match="candidate retrieval evaluation failed"):
        _pipeline(store, Source(changed, source.content), vectors, evaluator).run()

    assert store.active_generation() == first.generation
    assert store.events == [("publish", first.generation)]
    assert {entry.record.metadata["vector_generation"] for entry in vectors.records} == {first.generation}