"""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}