diff --git a/harness/tests/test_evidence_formula_migration.py b/harness/tests/test_evidence_formula_migration.py index dd266c3f..743498df 100644 --- a/harness/tests/test_evidence_formula_migration.py +++ b/harness/tests/test_evidence_formula_migration.py @@ -24,3 +24,21 @@ def test_reviewed_formula_migration_has_deterministic_provenance_hash(): assert second is not None assert first.provenance.source_sha256 == second.provenance.source_sha256 assert first.provenance.source_sha256.startswith("sha256:") + + +def test_incompatible_reviewed_legacy_formula_fails_closed_with_the_original_record(): + formula = ConceptFormula( + concept="ablazione", + columns=["testo"], + sql="SELECT 2", + status="reviewed", + sources=["legacy manual"], + ) + + outcome = formula_store.legacy_formula_to_curated( + formula, legacy_path="formulas/ablazione-2.sql.md", + ) + + assert outcome.code == "legacy_formula_requires_manual_review" + assert outcome.legacy_path == "formulas/ablazione-2.sql.md" + assert outcome.formula == formula diff --git a/harness/tests/test_formula.py b/harness/tests/test_formula.py index 700809c5..3db2f518 100644 --- a/harness/tests/test_formula.py +++ b/harness/tests/test_formula.py @@ -104,7 +104,7 @@ def test_reviewed_legacy_formula_becomes_curated_formula_with_stable_provenance( ) assert migrated is not None - assert migrated.id == "evidence:fascia-pediatrica" + assert migrated.id.startswith("evidence:fascia-pediatrica-") assert migrated.title == "Fascia pediatrica" assert migrated.kind == "formula" assert migrated.payload.concept == formula.concept @@ -118,6 +118,34 @@ def test_reviewed_legacy_formula_becomes_curated_formula_with_stable_provenance( ).id == migrated.id +def test_reviewed_legacy_formulas_with_the_same_concept_keep_distinct_path_identities(): + first = ConceptFormula( + concept="fascia pediatrica", + columns=["clinical.patient.birth_date"], + sql="CASE WHEN age < 18 THEN 'pediatrica' ELSE 'adulta' END", + status="reviewed", + ) + second = ConceptFormula( + concept="fascia pediatrica", + columns=["clinical.patient.birth_date"], + sql="CASE WHEN age < 16 THEN 'pediatrica' ELSE 'adulta' END", + status="reviewed", + ) + + first_migration = formula_store.legacy_formula_to_curated( + first, legacy_path="formulas/fascia-pediatrica-1.sql.md", + ) + second_migration = formula_store.legacy_formula_to_curated( + second, legacy_path="formulas/fascia-pediatrica-2.sql.md", + ) + + assert first_migration is not None + assert second_migration is not None + assert first_migration.id != second_migration.id + assert first_migration.id.startswith("evidence:fascia-pediatrica-") + assert second_migration.id.startswith("evidence:fascia-pediatrica-") + + def test_session_formula_proposal_is_versioned_and_is_not_published_evidence(tmp_path): linking = SchemaLinking( question="Conta i pazienti pediatrici", @@ -148,3 +176,30 @@ def test_session_formula_proposal_is_versioned_and_is_not_published_evidence(tmp "sources": ["Sintetizzata nella sessione"], "decision_seq": 9, }] + + +def test_session_formula_proposals_require_an_unretracted_positive_f4_decision(tmp_path): + linking = SchemaLinking( + question="Conta i pazienti pediatrici", + concept_formulas=[ + {"concept": "approvata", "columns": [], "sql": "1", "sources": []}, + {"concept": "rifiutata", "columns": [], "sql": "2", "sources": []}, + {"concept": "indecisa", "columns": [], "sql": "3", "sources": []}, + ], + ) + decisions = [ + DecisionRecord( + seq=3, ts=datetime(2026, 8, 25, tzinfo=UTC), type="concept_formula_approved", + subject="phase:4", detail="approvata", + ), + DecisionRecord( + seq=4, ts=datetime(2026, 8, 25, tzinfo=UTC), type="concept_formula_rejected", + subject="phase:4", detail="rifiutata", + ), + ] + + projected = project_session(decisions, linking, tmp_path / "evidence") + + assert [(proposal["concept"], proposal["decision_seq"]) for proposal in projected] == [ + ("approvata", 3), + ] diff --git a/harness/tht/evidence/formula_store.py b/harness/tht/evidence/formula_store.py index cd020fc0..4e787279 100644 --- a/harness/tht/evidence/formula_store.py +++ b/harness/tht/evidence/formula_store.py @@ -8,11 +8,12 @@ from __future__ import annotations import hashlib import re +from dataclasses import dataclass from pathlib import Path from typing import Literal import yaml -from pydantic import BaseModel +from pydantic import BaseModel, ValidationError from tht.evidence.canonical import CuratedEvidence @@ -56,6 +57,16 @@ class ConceptFormula(BaseModel): return cls.model_validate({**meta, "sql": body.strip("\n")}) +@dataclass(frozen=True) +class LegacyFormulaMigrationFailure: + """A reviewed legacy formula that must be resolved manually before publication.""" + + code: Literal["legacy_formula_requires_manual_review"] + legacy_path: str + formula: ConceptFormula + problems: tuple[str, ...] + + def _next_path(root: Path, slug: str) -> Path: """First free -.sql.md path under root (n starts at 1).""" root.mkdir(parents=True, exist_ok=True) @@ -110,7 +121,7 @@ def legacy_formula_to_curated( formula: ConceptFormula, *, legacy_path: str, -) -> CuratedEvidence | None: +) -> CuratedEvidence | LegacyFormulaMigrationFailure | None: """Convert one reviewed legacy formula into its deterministic curated counterpart. Drafts and model-generated formulas have no global publication status. Their @@ -123,23 +134,38 @@ def legacy_formula_to_curated( ) source_sha256 = hashlib.sha256(formula.dump().encode("utf-8")).hexdigest() source_file = legacy_path if legacy_path.startswith("source/") else f"source/{legacy_path}" - return CuratedEvidence.model_validate({ - "schema_version": 1, - "id": f"evidence:{formula._slug}", - "title": formula.concept[:1].upper() + formula.concept[1:], - "kind": "formula", - "purposes": ["schema_linking", "sql_generation"], - "applies_to": {"concepts": [formula.concept], "columns": formula.columns}, - "language": "it", - "provenance": { - "source_file": source_file, - "source_sha256": f"sha256:{source_sha256}", - "supporting_excerpts": source_notes, - }, - "review_items": [], - "payload": { - "concept": formula.concept, - "columns": formula.columns, - "sql": formula.sql, - }, - }) + # The legacy path is the immutable identity of this unit during migration. Keeping + # its full digest avoids a duplicate public ID when the same concept has reviewed + # competing formulas, while retaining the readable concept slug as the prefix. + legacy_identity = hashlib.sha256(legacy_path.encode("utf-8")).hexdigest() + try: + return CuratedEvidence.model_validate({ + "schema_version": 1, + "id": f"evidence:{formula._slug}-{legacy_identity}", + "title": formula.concept[:1].upper() + formula.concept[1:], + "kind": "formula", + "purposes": ["schema_linking", "sql_generation"], + "applies_to": {"concepts": [formula.concept], "columns": formula.columns}, + "language": "it", + "provenance": { + "source_file": source_file, + "source_sha256": f"sha256:{source_sha256}", + "supporting_excerpts": source_notes, + }, + "review_items": [], + "payload": { + "concept": formula.concept, + "columns": formula.columns, + "sql": formula.sql, + }, + }) + except ValidationError as error: + return LegacyFormulaMigrationFailure( + code="legacy_formula_requires_manual_review", + legacy_path=legacy_path, + formula=formula, + problems=tuple(sorted( + ".".join(str(part) for part in issue["loc"]) + for issue in error.errors() + )), + ) diff --git a/harness/tht/evidence/session.py b/harness/tht/evidence/session.py index d74d35f0..8131c8b5 100644 --- a/harness/tht/evidence/session.py +++ b/harness/tht/evidence/session.py @@ -64,11 +64,17 @@ def _formula_proposals(decisions: list["DecisionRecord"], linking: "SchemaLinkin A proposal remains in the persisted session artifact until a separate curator imports, reviews, and publishes it in the workspace repository. """ - approved = { - decision.detail: decision.seq + retracted = { + decision.retracts for decision in decisions - if decision.type == "concept_formula_approved" and decision.detail + if decision.type == "decision_retracted" and decision.retracts is not None } + latest_f4_decision = {} + for decision in sorted(decisions, key=lambda item: item.seq): + if decision.seq in retracted or decision.subject != "phase:4" or not decision.detail: + continue + if decision.type in {"concept_formula_approved", "concept_formula_rejected"}: + latest_f4_decision[decision.detail] = decision proposals = [] for formula in linking.concept_formulas: if not isinstance(formula, dict): @@ -83,6 +89,9 @@ def _formula_proposals(decisions: list["DecisionRecord"], linking: "SchemaLinkin evidence_id = formula.get("evidence_id", formula.get("id", "")) if isinstance(evidence_id, str) and evidence_id.startswith("evidence:"): continue + decision = latest_f4_decision.get(concept) + if decision is None or decision.type != "concept_formula_approved": + continue proposals.append({ "schema_version": 1, "kind": "formula_proposal", @@ -91,7 +100,7 @@ def _formula_proposals(decisions: list["DecisionRecord"], linking: "SchemaLinkin "columns": columns, "sql": sql, "sources": formula.get("sources", []), - "decision_seq": approved.get(concept), + "decision_seq": decision.seq, }) return proposals