diff --git a/harness/nsp/decisions.py b/harness/nsp/decisions.py index 7c7de0a7..496fb0c6 100644 --- a/harness/nsp/decisions.py +++ b/harness/nsp/decisions.py @@ -39,6 +39,11 @@ DecisionType = Literal[ # "phase:4", detail = il valore (es. "ablazione"), rationale = la/e colonna/e scelta/e # dal reviewer (aggregate_lsh_multi le espone tutte senza collassare al miglior match). "value_grounded", + # D14b: formula di concetto approvata/rifiutata dal reviewer. subject = "phase:4", + # detail = il concetto (es. "fascia pediatrica"), rationale = la/e colonna/e o il motivo. + # retrieve_formula restituisce i candidati; queste decisioni registrano la scelta. + "concept_formula_approved", + "concept_formula_rejected", ] diff --git a/harness/nsp/evidence/formula_store.py b/harness/nsp/evidence/formula_store.py new file mode 100644 index 00000000..47c8117e --- /dev/null +++ b/harness/nsp/evidence/formula_store.py @@ -0,0 +1,100 @@ +"""SQL concept->formula evidence store (spec D14b, ยง4.7). + +A concept (e.g. 'fascia pediatrica', 'ablazione') maps to a reusable SQL formula +(a CASE WHEN ...) that derives it from physical columns. These are reviewable +units: the gate surfaces a candidate formula, the reviewer approves or rejects it +(decision types concept_formula_approved / concept_formula_rejected), and approved +formulas travel with the schema-linking artifact. + +Storage: one file per formula, frontmatter YAML + SQL body (same shape as +EvidenceDoc.parse). Directory layout: /formulas/-.sql.md. +Retrieve is by concept (may return several, e.g. competing drafts vs reviewed). +""" +from __future__ import annotations + +import re +from pathlib import Path +from typing import Literal + +import yaml +from pydantic import BaseModel + +FORMULAS_SUBDIR = "formulas" +_SUFFIX_RE = re.compile(r"^(.*?)-(\d+)\.sql\.md$") + + +class ConceptFormula(BaseModel): + concept: str + columns: list[str] = [] + sql: str + status: Literal["draft", "reviewed"] = "draft" + sources: list[str] = [] + + @property + def _slug(self) -> str: + """ASCII slug for the filename (matches textutil.slugify shape).""" + import unicodedata + + text = unicodedata.normalize("NFKD", self.concept).encode("ascii", "ignore").decode() + return re.sub(r"[^a-z0-9_]+", "-", text.lower()).strip("-") or "formula" + + def dump(self) -> str: + meta = self.model_dump(exclude={"sql"}, mode="json") + fm = yaml.safe_dump(meta, sort_keys=False, allow_unicode=True) + return f"---\n{fm}---\n{self.sql}\n" + + @classmethod + def parse(cls, text: str) -> "ConceptFormula": + if not text.startswith("---\n"): + raise ValueError("frontmatter mancante (atteso '---\\n' iniziale)") + try: + _, fm, body = text.split("---\n", 2) + except ValueError as e: + raise ValueError("frontmatter malformato") from e + meta = yaml.safe_load(fm) + if not isinstance(meta, dict): + raise ValueError("frontmatter non valido") + return cls.model_validate({**meta, "sql": body.strip("\n")}) + + +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) + existing = sorted(root.glob(f"{slug}-*.sql.md")) + n = 0 + for p in existing: + m = _SUFFIX_RE.match(p.name) + if m: + n = max(n, int(m.group(2))) + return root / f"{slug}-{n + 1}.sql.md" + + +def save_formula(root: Path | str, formula: ConceptFormula) -> Path: + """Persist a single concept->formula unit under /formulas/. Returns the + written path. Append-only: each save writes a new file (so competing drafts and + reviewed versions coexist until a curator prunes).""" + root = Path(root) + formulas_dir = root / FORMULAS_SUBDIR + path = _next_path(formulas_dir, formula._slug) + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(formula.dump()) + return path + + +def retrieve_formula(root: Path | str, concept: str) -> list[ConceptFormula]: + """All formulas for `concept` under /formulas/. Empty list if none (or if + the dir is absent). Multiple results mean competing drafts/versions for the same + concept -- the caller (gate) lets the reviewer pick.""" + root = Path(root) + formulas_dir = root / FORMULAS_SUBDIR + if not formulas_dir.is_dir(): + return [] + out: list[ConceptFormula] = [] + for f in sorted(formulas_dir.glob("*.sql.md")): + try: + formula = ConceptFormula.parse(f.read_text()) + except ValueError: + continue # malformed file: skip, don't crash retrieval + if formula.concept == concept: + out.append(formula) + return out diff --git a/harness/tests/test_formula.py b/harness/tests/test_formula.py new file mode 100644 index 00000000..10eb6370 --- /dev/null +++ b/harness/tests/test_formula.py @@ -0,0 +1,85 @@ +"""L1: SQL formula evidence -- concept->formula units (spec D14b). + +A concept (e.g. 'fascia pediatrica') maps to a SQL formula (CASE WHEN ...) that +derives it from physical columns. These are reusable, reviewable units: the gate +surfaces a candidate formula, the reviewer approves or rejects it (recorded via +concept_formula_approved / concept_formula_rejected), and approved formulas are +part of the schema-linking artifact. The store is frontmatter-YAML + SQL body. +""" +from pathlib import Path + +import pytest + +from nsp.evidence.formula_store import ConceptFormula, retrieve_formula, save_formula + + +def test_formula_retrieval_by_concept(tmp_path): + f = ConceptFormula( + concept="fascia pediatrica", + columns=["data_nascita"], + sql="CASE WHEN age < 18 THEN 'pediatrica' ELSE 'adulto' END", + status="reviewed", + sources=["src1"], + ) + save_formula(tmp_path, f) + results = retrieve_formula(tmp_path, "fascia pediatrica") + assert len(results) == 1 + assert results[0].sql.startswith("CASE WHEN") + assert results[0].concept == "fascia pediatrica" + assert results[0].status == "reviewed" + assert "data_nascita" in results[0].columns + + +def test_save_and_reload_roundtrip_preserves_sql_body(tmp_path): + f = ConceptFormula( + concept="fascia pediatrica", + columns=["data_nascita"], + sql="CASE\n WHEN x THEN 1\nEND", + status="draft", + sources=[], + ) + path = save_formula(tmp_path, f) + assert path.exists() + # the file is frontmatter YAML + SQL body + text = path.read_text() + assert text.startswith("---") + assert "concept: fascia pediatrica" in text + assert "CASE" in text # SQL body preserved + + +def test_retrieve_multiple_formulas_for_concept(tmp_path): + # two competing formulas for the same concept (different sources/status) + save_formula(tmp_path, ConceptFormula(concept="ablazione", columns=["flag"], + sql="SELECT 1", status="draft", sources=["a"])) + save_formula(tmp_path, ConceptFormula(concept="ablazione", columns=["testo"], + sql="SELECT 2", status="reviewed", sources=["b"])) + results = retrieve_formula(tmp_path, "ablazione") + assert len(results) == 2 + statuses = {r.status for r in results} + assert statuses == {"draft", "reviewed"} + + +def test_retrieve_empty_when_no_match(tmp_path): + save_formula(tmp_path, ConceptFormula(concept="altro", columns=["c"], + sql="SELECT 1", status="reviewed", sources=[])) + assert retrieve_formula(tmp_path, "inesistente") == [] + + +def test_retrieve_empty_on_missing_dir(tmp_path): + # no formulas dir at all -> empty list, not error + assert retrieve_formula(tmp_path / "nope", "anything") == [] + + +def test_concept_formula_decision_types_exist(): + import typing + from nsp.decisions import DecisionType + args = typing.get_args(DecisionType) + assert "concept_formula_approved" in args + assert "concept_formula_rejected" in args + + +def test_concept_formula_default_status(tmp_path): + # status has a sensible default so an author can write a draft quickly + f = ConceptFormula(concept="x", columns=["c"], sql="SELECT 1") + assert f.status == "draft" # not yet reviewed + assert f.sources == []