feat(harness): SQL formula evidence -- concept->formula units + retrieval (D14b)
New evidence/formula_store.py: ConceptFormula (concept, columns, sql, status, sources) as a frontmatter-YAML + SQL-body unit, stored one-file-per-formula under <formulas>/<slug>-<n>.sql.md. save_formula is append-only (competing drafts and reviewed versions coexist); retrieve_formula(concept) returns all of them so the gate can surface candidates and let the reviewer choose. concept_formula_approved / concept_formula_rejected added to DecisionType (records the reviewer's choice; approved formulas travel with schema-linking). L1: test_formula (7 tests) -- retrieval by concept, save/reload roundtrip (SQL body preserved, frontmatter well-formed), multiple formulas per concept, empty on no-match / missing dir, decision-type existence, default draft status. Deferred: --kind formula on nsp search (needs search_cmd porting) wires retrieve_formula into the CLI; lands with the search command.
This commit is contained in:
@@ -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",
|
||||
]
|
||||
|
||||
|
||||
|
||||
@@ -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: <root>/formulas/<slug>-<n>.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 <slug>-<n>.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 <root>/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 <root>/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
|
||||
@@ -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 == []
|
||||
Reference in New Issue
Block a user