Codex
82e2c91f42
Publish documentation / publish (push) Successful in 1m27s
Add PostgreSQL-backed memory, editable evidence with source review and activation, and human-approved archive repairs across the harness, API, and UI. Include migrations, deployment support, regression coverage, and validation documentation. Refresh permissions from validated session roles so existing administrator logins can access newly deployed archive management features.
452 lines
15 KiB
Python
452 lines
15 KiB
Python
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),
|
|
]
|