zettelkasten.data_grounding

zettelkasten.data_grounding

Data-grounded evidence: verify a claim by re-computing a statistic.

This is the generalist substrate described in dev/documents/260707_report_data_grounded_evidence.md §3 (derivations + registry + the expr hatch) and §4 (verification, hard vs soft failure modes). It adds a THIRD grounding modality alongside quote (verbatim substring, see :mod:zettelkasten.grounding) and equation (snapshot): a claim can be backed by a re-computable statistic over a stored dataset, and the auditor confirms it by re-running the computation rather than string-matching.

angelo owns the evidence representation, verification, and reporting. It never interprets what a derivation means, never composes beliefs — a derivation is an opaque, registered, deterministic function angelo can only invoke and compare (§1). Two ways to name a derivation:

• expr hatch — a whitelisted mini-evaluator (mean:col / quantile:col:0.9 / sum:col …) that re-verifies anywhere with no host registration, so every descriptive (measured) claim is portable (§3.3, §4).
• registered derivation — host code (e.g. Freischutz) contributed via :func:register_derivation, same trust boundary as .cursor/agents.yaml (§3.1, §3.3). Fitted (inferred) claims live here.

Safety: the expr path NEVER uses eval/exec — it splits on : and dispatches to an allow-list of reducers, with no attribute access, calls, or imports. An unrecognized form or unknown column raises a clear ValueError.

Verification splits outcomes into two tiers (§4):

• hard violation — angelo could re-run and it disagreed (dataset_hash_mismatch, value_not_reproduced, nondeterministic_unpinned, validation_malformed). Blocks coverage like an unverified quote.
• soft unavailable — angelo cannot re-run here (derivation_unavailable, dataset_unavailable). Informational; the prior verified/verified_in stamp is retained so a once-verified claim does not read as broken merely because you are browsing a repo where its derivation isn't installed.

:func:verify_data_grounding never raises for an ordinary failure — it returns the structured outcome. It only lets truly unexpected internal errors propagate.

DerivationResult dataclass

The return contract of a derivation (§3.1).

Only value is required and verified. validation / confidence / method_meta are opaque — angelo stores and surfaces them but never computes with them (§2.2).

Source code in zettelkasten/data_grounding.py

@dataclasses.dataclass
class DerivationResult:
    """The return contract of a derivation (§3.1).

    Only ``value`` is required and verified. ``validation`` / ``confidence`` /
    ``method_meta`` are **opaque** — angelo stores and surfaces them but never
    computes with them (§2.2).
    """

    value: float
    validation: dict[str, Any] | None = None
    confidence: float | None = None
    method_meta: dict[str, Any] | None = None

RegisteredDerivation dataclass

A host-contributed derivation plus the metadata angelo verifies against.

Source code in zettelkasten/data_grounding.py

@dataclasses.dataclass
class RegisteredDerivation:
    """A host-contributed derivation plus the metadata angelo verifies against."""

    fn: Callable[[Any, dict, dict], DerivationResult]
    id: str
    version: str
    fit: bool = False
    deterministic: bool = True

DerivationLookupError

Bases: LookupError

Raised when (id, version) is not registered in this environment.

Distinct so callers can map it to the SOFT derivation_unavailable outcome rather than a hard violation (§4).

Source code in zettelkasten/data_grounding.py

class DerivationLookupError(LookupError):
    """Raised when ``(id, version)`` is not registered in this environment.

    Distinct so callers can map it to the SOFT ``derivation_unavailable`` outcome
    rather than a hard violation (§4).
    """

register_derivation

register_derivation(id: str, version: Any, *, fit: bool = False, deterministic: bool = True) -> Callable[[Callable[..., DerivationResult]], Callable[..., DerivationResult]]

Decorator registering a derivation callable keyed by (id, version) (§3.1).

The wrapped callable must have signature (table, selection, params) -> DerivationResult. fit distinguishes a fitted (inferred) derivation from a descriptive (measured) one; deterministic=False marks a stochastic derivation that must pin an RNG seed in params (§3.2).

Source code in zettelkasten/data_grounding.py

def register_derivation(
    id: str,
    version: Any,
    *,
    fit: bool = False,
    deterministic: bool = True,
) -> Callable[[Callable[..., DerivationResult]], Callable[..., DerivationResult]]:
    """Decorator registering a derivation callable keyed by ``(id, version)`` (§3.1).

    The wrapped callable must have signature ``(table, selection, params) ->
    DerivationResult``. ``fit`` distinguishes a fitted (``inferred``) derivation
    from a descriptive (``measured``) one; ``deterministic=False`` marks a
    stochastic derivation that must pin an RNG seed in ``params`` (§3.2).
    """

    def _decorator(fn: Callable[..., DerivationResult]) -> Callable[..., DerivationResult]:
        _REGISTRY[_key(id, version)] = RegisteredDerivation(
            fn=fn,
            id=str(id),
            version=str(version),
            fit=bool(fit),
            deterministic=bool(deterministic),
        )
        return fn

    return _decorator

get_derivation

get_derivation(id: str, version: Any) -> RegisteredDerivation

Look up a registered derivation, raising :class:DerivationLookupError.

Source code in zettelkasten/data_grounding.py

def get_derivation(id: str, version: Any) -> RegisteredDerivation:
    """Look up a registered derivation, raising :class:`DerivationLookupError`."""
    try:
        return _REGISTRY[_key(id, version)]
    except KeyError as exc:
        raise DerivationLookupError(
            f"derivation {id!r} version {version!r} is not registered here"
        ) from exc

eval_expr

eval_expr(expr: str, table: Any) -> float

Evaluate a whitelisted expression over a resolved dataset table (§3.3).

Supported forms (SAFE — split on : and dispatch, never eval/exec, no attribute access, no calls, no imports)::

mean:col   median:col   sum:col   std:col   count:col   min:col   max:col
quantile:col:<q>          # q in [0, 1], linear interpolation (type 7)

table is a :class:zettelkasten.datasets.DatasetTable (anything exposing columns and rows). Numeric reducers coerce cells to float and drop missing/non-numeric cells; count counts non-missing cells of any dtype. Raises ValueError for an unrecognized form or an unknown column.

LIMITATION (report item 10): because the form is split on :, a column whose NAME contains a : cannot be addressed through this hatch (the split would mis-parse it as reducer:col:extra). Such a column must be renamed, or grounded via a registered derivation (which receives the raw table and selects columns itself) instead of the expr hatch.

Source code in zettelkasten/data_grounding.py

def eval_expr(expr: str, table: Any) -> float:
    """Evaluate a whitelisted expression over a resolved dataset table (§3.3).

    Supported forms (SAFE — split on ``:`` and dispatch, never ``eval``/``exec``,
    no attribute access, no calls, no imports)::

        mean:col   median:col   sum:col   std:col   count:col   min:col   max:col
        quantile:col:<q>          # q in [0, 1], linear interpolation (type 7)

    ``table`` is a :class:`zettelkasten.datasets.DatasetTable` (anything exposing
    ``columns`` and ``rows``). Numeric reducers coerce cells to ``float`` and drop
    missing/non-numeric cells; ``count`` counts non-missing cells of any dtype.
    Raises ``ValueError`` for an unrecognized form or an unknown column.

    LIMITATION (report item 10): because the form is split on ``:``, a column
    whose NAME contains a ``:`` cannot be addressed through this hatch (the split
    would mis-parse it as ``reducer:col:extra``). Such a column must be renamed,
    or grounded via a registered derivation (which receives the raw table and
    selects columns itself) instead of the ``expr`` hatch.
    """
    if not isinstance(expr, str) or ":" not in expr:
        raise ValueError(f"not an expression form (expected 'name:col'): {expr!r}")

    parts = expr.split(":")
    name = parts[0].strip().casefold()

    if name == "quantile":
        if len(parts) != 3:
            raise ValueError(
                f"quantile expects 'quantile:col:<q>', got {expr!r}"
            )
        idx = _column_index(table, parts[1])
        try:
            q = float(parts[2].strip())
        except (TypeError, ValueError) as exc:
            raise ValueError(f"quantile q must be a number in [0, 1]: {parts[2]!r}") from exc
        if not 0.0 <= q <= 1.0:
            raise ValueError(f"quantile q must be in [0, 1], got {q}")
        values = [f for f in (_to_float(c) for c in _column_cells(table, idx)) if f is not None]
        return _quantile(values, q)

    if len(parts) != 2 or name not in _SCALAR_REDUCERS:
        raise ValueError(
            f"unrecognized expression {expr!r}; allowed: "
            f"{sorted(_SCALAR_REDUCERS)} + 'quantile:col:<q>'"
        )

    idx = _column_index(table, parts[1])
    cells = _column_cells(table, idx)
    if name == "count":
        return float(sum(1 for c in cells if not _is_missing(c)))
    values = [f for f in (_to_float(c) for c in cells) if f is not None]
    return _reduce(name, values)

apply_selection

apply_selection(table: Any, selection: dict | None) -> Any

Apply a simple, deterministic where/columns filter (§4).

selection may be None/empty (identity) or a dict with:

• where — a predicate string; a conjunction of simple comparisons joined by and (case-insensitive), each col <op> literal with op one of >= <= == != > < =. This is deliberately minimal so descriptive claims re-verify anywhere; a richer predicate belongs in a registered derivation (which receives the raw table + selection and applies its own filter).
• columns — an optional list restricting/ordering columns.

Any other key raises ValueError (an unrecognized selection is a broken pin, not silently ignored).

Source code in zettelkasten/data_grounding.py

def apply_selection(table: Any, selection: dict | None) -> Any:
    """Apply a simple, deterministic ``where``/``columns`` filter (§4).

    ``selection`` may be ``None``/empty (identity) or a dict with:

    * ``where`` — a predicate string; a conjunction of simple comparisons joined
      by ``and`` (case-insensitive), each ``col <op> literal`` with ``op`` one of
      ``>= <= == != > < =``. This is deliberately minimal so descriptive claims
      re-verify anywhere; a richer predicate belongs in a registered derivation
      (which receives the raw table + selection and applies its own filter).
    * ``columns`` — an optional list restricting/ordering columns.

    Any other key raises ``ValueError`` (an unrecognized selection is a broken
    pin, not silently ignored).
    """
    if not selection:
        return table
    if not isinstance(selection, dict):
        raise ValueError(f"selection must be a mapping, got {type(selection).__name__}")

    unknown = set(selection) - {"where", "columns"}
    if unknown:
        raise ValueError(f"unsupported selection key(s): {sorted(unknown)}")

    columns = list(getattr(table, "columns", []) or [])
    rows = [list(r) for r in (getattr(table, "rows", []) or [])]

    where = selection.get("where")
    if where:
        if not isinstance(where, str):
            raise ValueError(f"selection.where must be a string, got {type(where).__name__}")
        clauses = _split_conjunction(where)
        parsed = [_parse_clause(c) for c in clauses]
        col_idx = {str(c): i for i, c in enumerate(columns)}
        for col, _, _ in parsed:
            if col not in col_idx:
                raise ValueError(f"unknown column {col!r} in where clause")
        kept: list[list[Any]] = []
        for row in rows:
            ok = True
            for col, op, literal in parsed:
                i = col_idx[col]
                cell = row[i] if i < len(row) else None
                if not _compare(cell, op, literal):
                    ok = False
                    break
            if ok:
                kept.append(row)
        rows = kept

    wanted = selection.get("columns")
    if wanted:
        if not isinstance(wanted, (list, tuple)):
            raise ValueError("selection.columns must be a list")
        keep = [str(c) for c in wanted]
        col_idx = {str(c): i for i, c in enumerate(columns)}
        missing = [c for c in keep if c not in col_idx]
        if missing:
            raise ValueError(f"unknown column(s) in selection.columns: {missing}")
        idxs = [col_idx[c] for c in keep]
        columns = keep
        rows = [[row[i] if i < len(row) else None for i in idxs] for row in rows]

    return dataclasses.replace(table, columns=columns, rows=rows, num_rows=len(rows))

table_content_hash

table_content_hash(table: Any) -> str

Deterministic sha256: hash identifying a resolved table's contents.

Canonicalization (order-independent, so a benign row/column reordering of the SAME data hashes identically): columns are sorted; each row's cells are permuted to that sorted column order and normalized via :func:_canon_cell (consistent float formatting, explicit NaN/null tokens); the normalized row strings are then sorted. The digest covers the sorted column header plus the sorted normalized rows.

Source code in zettelkasten/data_grounding.py

def table_content_hash(table: Any) -> str:
    """Deterministic ``sha256:`` hash identifying a resolved table's contents.

    Canonicalization (order-independent, so a benign row/column reordering of the
    SAME data hashes identically): columns are sorted; each row's cells are
    permuted to that sorted column order and normalized via :func:`_canon_cell`
    (consistent float formatting, explicit NaN/null tokens); the normalized row
    strings are then sorted. The digest covers the sorted column header plus the
    sorted normalized rows.
    """
    columns = [str(c) for c in (getattr(table, "columns", []) or [])]
    order = sorted(range(len(columns)), key=lambda i: columns[i])
    sorted_cols = [columns[i] for i in order]

    row_strings: list[str] = []
    for row in getattr(table, "rows", []) or []:
        cells = [row[i] if i < len(row) else None for i in order]
        row_strings.append("\x1f".join(_canon_cell(c) for c in cells))
    row_strings.sort()

    hasher = hashlib.sha256()
    hasher.update(("\x1e".join(sorted_cols) + "\x1d").encode("utf-8"))
    for rs in row_strings:
        hasher.update(rs.encode("utf-8"))
        hasher.update(b"\x1e")
    return "sha256:" + hasher.hexdigest()

verify_data_grounding

verify_data_grounding(note: Any, graph_dir: Path | str) -> dict[str, Any]

Re-run a method: data grounding and report the structured outcome (§4).

note is a claim :class:zettelkasten.graph.Note whose grounding is a method: data citation; graph_dir is the source-folder path the note lives in (ZettelGraph.path). Never raises for an ordinary failure — every hard violation and soft unavailability is returned as a structured dict. Only a truly unexpected internal bug propagates.

Steps: validate validation shape → re-resolve the dataset note → content-hash the resolved table and compare to the pin → resolve the derivation (expr hatch or registry) → re-run and compare value within tolerance. A clean reproduce returns {verified: True, value: <recomputed>, ...}.

Source code in zettelkasten/data_grounding.py

def verify_data_grounding(note: Any, graph_dir: Path | str) -> dict[str, Any]:
    """Re-run a ``method: data`` grounding and report the structured outcome (§4).

    ``note`` is a claim :class:`zettelkasten.graph.Note` whose ``grounding`` is a
    ``method: data`` citation; ``graph_dir`` is the source-folder path the note
    lives in (``ZettelGraph.path``). Never raises for an ordinary failure — every
    hard violation and soft unavailability is returned as a structured dict. Only
    a truly unexpected internal bug propagates.

    Steps: validate ``validation`` shape → re-resolve the dataset note →
    content-hash the resolved table and compare to the pin → resolve the
    derivation (``expr`` hatch or registry) → re-run and compare ``value`` within
    ``tolerance``. A clean reproduce returns ``{verified: True, value: <recomputed>,
    ...}``.
    """
    grounding = getattr(note, "grounding", None)
    if not isinstance(grounding, dict):
        raise ValueError("verify_data_grounding requires a note with a grounding dict")

    graph_dir = Path(graph_dir)

    # (1) Structural: a present `validation` must be a {kind, ...} dict (§4). This
    # is environment-independent, so it is checked before availability.
    validation = grounding.get("validation")
    if validation is not None and (not isinstance(validation, dict) or "kind" not in validation):
        return _outcome(
            verified=False,
            severity="hard",
            failure_mode="validation_malformed",
            reason="`validation` must be a dict with a `kind` key.",
            grounding=grounding,
        )

    # (1b) Structural: the tolerances must be numeric if present (also
    # environment-independent). A non-numeric ``tolerance``/``rel_tolerance`` is a
    # broken pin — surface it as a HARD violation rather than letting a bare
    # ``float()`` raise and abort the surrounding extraction audit.
    if (
        _tol_or_none(grounding.get("tolerance"), DEFAULT_TOLERANCE) is None
        or _tol_or_none(grounding.get("rel_tolerance"), 0.0) is None
    ):
        return _outcome(
            verified=False,
            severity="hard",
            failure_mode="validation_malformed",
            reason="`tolerance`/`rel_tolerance` must be numeric.",
            grounding=grounding,
        )

    # (2) Re-resolve the dataset note. Missing here => SOFT dataset_unavailable.
    loaded = _load_dataset_note(
        grounding.get("dataset_note", ""), graph_dir, grounding.get("dataset_graph", "")
    )
    if loaded is None:
        return _outcome(
            verified=bool(grounding.get("verified", False)),
            severity="soft",
            failure_mode="dataset_unavailable",
            reason=(
                f"dataset note {grounding.get('dataset_note')!r} is not resolvable here."
            ),
            grounding=grounding,
        )
    dataset_note, dataset_dir = loaded
    table = datasets_mod.resolve_dataset(dataset_note, dataset_dir)
    if not getattr(table, "actual_hash", ""):
        # resolve_dataset leaves actual_hash empty only when no bytes resolved.
        return _outcome(
            verified=bool(grounding.get("verified", False)),
            severity="soft",
            failure_mode="dataset_unavailable",
            reason=(
                "dataset resolved no bytes"
                + (f": {table.warning}" if getattr(table, "warning", "") else ".")
            ),
            grounding=grounding,
        )

    # (3) Content-hash the resolved table and compare to the pin (§4). Accept
    # either the canonical table hash (primary, per the task spec) OR the resolved
    # bytes hash (§8: the snapshot's content_hash) so verification is robust to
    # whichever scheme the writer pinned — genuine data drift changes both.
    #
    # CAVEAT (report item 9): the primary ``table_content_hash`` is
    # order-INDEPENDENT (rows and columns are sorted before hashing), so a benign
    # reordering of the SAME data hashes identically. That is correct for the
    # descriptive ``expr`` reducers (mean/sum/quantile are order-invariant) but is
    # UNSAFE for a registered derivation whose value is ORDER-DEPENDENT (e.g. a
    # time-series/backtest that consumes rows in sequence): a row permutation would
    # still match the pin yet could change the derived value. The acceptance below
    # deliberately keeps the existing multi-scheme match for backward-compatibility
    # with already-pinned claims; an order-sensitive derivation should additionally
    # pin the bytes ``content_hash`` (order-preserving) rather than rely on the
    # canonical table hash alone.
    computed_hash = table_content_hash(table)
    declared_hash = str(grounding.get("dataset_hash", "") or "")
    if not declared_hash:
        # An unpinned data citation would verify against whatever bytes are on
        # disk right now, so the drift guard cannot fire — treat a missing/empty
        # dataset_hash as a HARD violation instead of silently skipping the check.
        return _outcome(
            verified=False,
            severity="hard",
            failure_mode="unpinned_dataset",
            reason=(
                "grounding pins no `dataset_hash`; an unpinned data citation "
                "verifies against the current on-disk bytes and cannot detect "
                "drift. Pin the dataset_hash of the table the value was computed "
                "over."
            ),
            grounding=grounding,
            hash_actual=computed_hash,
            hash_match=None,
        )
    declared_norm = datasets_mod._normalize_hash(declared_hash)
    candidates = {
        datasets_mod._normalize_hash(computed_hash),
        datasets_mod._normalize_hash(getattr(table, "actual_hash", "") or ""),
        datasets_mod._normalize_hash(getattr(table, "content_hash", "") or ""),
    }
    hash_match: bool | None = declared_norm in {c for c in candidates if c}
    if not hash_match:
        return _outcome(
            verified=False,
            severity="hard",
            failure_mode="dataset_hash_mismatch",
            reason=(
                f"pinned dataset_hash {declared_norm[:12]}… does not match the "
                f"resolved table hash {datasets_mod._normalize_hash(computed_hash)[:12]}…."
            ),
            grounding=grounding,
            hash_expected=declared_hash,
            hash_actual=computed_hash,
            hash_match=False,
        )

    # (4) Resolve the derivation: expr hatch (contains ':') vs registry (§3, §4).
    derivation_id = str(grounding.get("derivation_id", "") or "")
    version = grounding.get("derivation_version")
    selection = grounding.get("selection")
    params = grounding.get("params") or {}
    is_expr = ":" in derivation_id
    deriv_meta: dict[str, Any] = {
        "id": derivation_id,
        "version": version,
        "kind": "expr" if is_expr else "registered",
    }

    if is_expr:
        try:
            filtered = apply_selection(table, selection if isinstance(selection, dict) else None)
            recomputed = eval_expr(derivation_id, filtered)
        except ValueError as exc:
            return _outcome(
                verified=False,
                severity="hard",
                failure_mode="value_not_reproduced",
                reason=f"expression could not be evaluated: {exc}",
                grounding=grounding,
                hash_expected=declared_hash,
                hash_actual=computed_hash,
                hash_match=hash_match,
                deriv_meta=deriv_meta,
            )
    else:
        try:
            registered = get_derivation(derivation_id, version)
        except DerivationLookupError as exc:
            # Distinguish a pure version drift (the id IS registered here, just not
            # the pinned version) from an entirely unknown derivation, so a stale
            # version pin surfaces its own soft code instead of reading as a
            # missing derivation.
            id_known = any(k[0] == str(derivation_id) for k in _REGISTRY)
            return _outcome(
                verified=bool(grounding.get("verified", False)),
                severity="soft",
                failure_mode=(
                    "derivation_version_unavailable" if id_known
                    else "derivation_unavailable"
                ),
                reason=str(exc),
                grounding=grounding,
                hash_expected=declared_hash,
                hash_actual=computed_hash,
                hash_match=hash_match,
                deriv_meta=deriv_meta,
            )
        deriv_meta.update({"fit": registered.fit, "deterministic": registered.deterministic})
        # A stochastic derivation must pin an RNG seed in params (§3.2).
        if not registered.deterministic and "seed" not in params:
            return _outcome(
                verified=False,
                severity="hard",
                failure_mode="nondeterministic_unpinned",
                reason=(
                    "derivation is registered deterministic=False but params pins no "
                    "`seed`; a stochastic derivation must pin its RNG seed (§3.2)."
                ),
                grounding=grounding,
                hash_expected=declared_hash,
                hash_actual=computed_hash,
                hash_match=hash_match,
                deriv_meta=deriv_meta,
            )
        # Registered derivations receive the RAW table + selection + params and
        # apply their own (host-powered) selection (§3.1).
        try:
            result = registered.fn(table, selection if isinstance(selection, dict) else {}, params)
            recomputed = float(result.value)
        except Exception as exc:  # host code failed to reproduce — a hard violation
            return _outcome(
                verified=False,
                severity="hard",
                failure_mode="value_not_reproduced",
                reason=f"derivation raised while re-running: {exc}",
                grounding=grounding,
                hash_expected=declared_hash,
                hash_actual=computed_hash,
                hash_match=hash_match,
                deriv_meta=deriv_meta,
            )

    # (5) Compare the re-run against the asserted value within tolerance (§4).
    # Tolerances were validated numeric above, so these never return None; an
    # explicit ``tolerance: 0`` is preserved (exact match) rather than defaulted.
    expected = grounding.get("value")
    abs_tol = _tol_or_none(grounding.get("tolerance"), DEFAULT_TOLERANCE)
    rel_tol = _tol_or_none(grounding.get("rel_tolerance"), 0.0)
    try:
        expected_f = float(expected)
    except (TypeError, ValueError):
        return _outcome(
            verified=False,
            severity="hard",
            failure_mode="value_not_reproduced",
            reason=f"grounding.value is not numeric: {expected!r}",
            grounding=grounding,
            recomputed=recomputed,
            hash_expected=declared_hash,
            hash_actual=computed_hash,
            hash_match=hash_match,
            deriv_meta=deriv_meta,
        )

    reproduces = not math.isnan(recomputed) and math.isclose(
        recomputed, expected_f, rel_tol=rel_tol, abs_tol=abs_tol
    )
    if not reproduces:
        return _outcome(
            verified=False,
            severity="hard",
            failure_mode="value_not_reproduced",
            reason=(
                f"re-run produced {recomputed!r}, which differs from the asserted "
                f"{expected_f!r} beyond tolerance (abs={abs_tol}, rel={rel_tol})."
            ),
            grounding=grounding,
            recomputed=recomputed,
            hash_expected=declared_hash,
            hash_actual=computed_hash,
            hash_match=hash_match,
            deriv_meta=deriv_meta,
        )

    return _outcome(
        verified=True,
        severity="ok",
        failure_mode=None,
        reason="re-computation reproduces the asserted value within tolerance.",
        grounding=grounding,
        recomputed=recomputed,
        hash_expected=declared_hash,
        hash_actual=computed_hash,
        hash_match=hash_match,
        deriv_meta=deriv_meta,
    )