ThothII/harness/tht/evidence/formula_store.py

"""Legacy ConceptFormula reader and one-way migration into Curated Evidence.

The ``formulas/*.sql.md`` store is retained only for the migration window. Runtime
lookup uses typed, published ``kind=formula`` Evidence instead. A session reviewer
may still approve a formula locally; that is a proposal, not publication.
"""
from __future__ import annotations

import hashlib
import re
import unicodedata
from dataclasses import dataclass
from pathlib import Path
from typing import Literal

import yaml
from pydantic import BaseModel, ValidationError

from tht.evidence.canonical import CuratedEvidence

FORMULAS_SUBDIR = "formulas"
_SUFFIX_RE = re.compile(r"^(.*?)-(\d+)\.sql\.md$")


class ConceptFormula(BaseModel):
    concept: str
    columns: list[str] = []
    sql: str
    # auto = sintetizzata dal modello (non ancora rivista); draft = bozza umana;
    # reviewed = approvata da un revisore. (spec §4.7.2: status auto/draft/reviewed)
    status: Literal["auto", "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 TypeError("frontmatter non valido")
        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 <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 _load_all(root: Path) -> list[ConceptFormula]:
    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:
            out.append(ConceptFormula.parse(f.read_text()))
        except ValueError:
            continue  # malformed file: skip, don't crash retrieval
    return out


def retrieve_formula(root: Path | str, concept: str) -> list[ConceptFormula]:
    """All formulas matching `concept` exactly 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."""
    return [f for f in _load_all(Path(root)) if f.concept == concept]


def search_formulas(root: Path | str, query: str) -> list[ConceptFormula]:
    """Read legacy formulas for migration tooling only (case-insensitive concept match)."""
    q = query.strip().lower()
    return [f for f in _load_all(Path(root)) if q in f.concept.lower()]


def legacy_formula_to_curated(
    formula: ConceptFormula,
    *,
    legacy_path: str,
    source_content: str | None = 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
    caller must project them as session-local Formula proposals instead.
    """
    if formula.status != "reviewed":
        return None
    problems: list[str] = []
    normalized_source: str | None = None
    if source_content is None:
        problems.append("original_source_required")
    else:
        from tht.evidence.authoring import normalize_source_text

        normalized_source = normalize_source_text(source_content)
        try:
            original_formula = ConceptFormula.parse(source_content)
        except (TypeError, ValidationError, ValueError, yaml.YAMLError):
            problems.append("original_source_invalid")
        else:
            if original_formula != formula:
                problems.append("original_source_mismatch")
    source_notes = tuple(formula.sources)
    if not source_notes:
        problems.append("supporting_excerpts_required")
    elif normalized_source is not None and any(
        unicodedata.normalize("NFC", note.replace("\r\n", "\n").replace("\r", "\n"))
        not in normalized_source
        for note in source_notes
    ):
        problems.append("supporting_excerpt_unverified")
    if problems:
        return LegacyFormulaMigrationFailure(
            code="legacy_formula_requires_manual_review",
            legacy_path=legacy_path,
            formula=formula,
            problems=tuple(sorted(problems)),
        )
    assert normalized_source is not None
    source_sha256 = hashlib.sha256(normalized_source.encode("utf-8")).hexdigest()
    source_file = legacy_path if legacy_path.startswith("source/") else f"source/{legacy_path}"
    # 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()
            )),
        )