diff --git a/CHANGELOG.md b/CHANGELOG.md index f753b26d..84c0dae1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -59,6 +59,21 @@ below corresponds to one such version. bounded to its 50-row sample on every engine through the new `Dialect.limited`, where it used to be bounded only where the row-limit keyword was `LIMIT`. (ACE-114) +- **`reconcile.py` reads any input a person brings, and grades a supplied statement part by part.** + Three new verbs beside `parse`, `diff` and `band`, which are unchanged. `intake` reads the four + shapes the reconcile skill will accept, a list of questions, questions with the SQL the person + trusts, labels with numbers, and labels with numbers and the SQL behind each tile, into one row + shape with the question, the statement and the expected value, any of which may be missing. A CSV + whose third column is SQL now yields a statement instead of a label with SQL glued onto it; a + statement whose label matches a tile joins that tile's row. `ledger` grades every part of a + supplied statement from the files the skill wrote beside it: what happened when it ran, what + `sm prepare` and `sm receipt` said, what `sm join-probes` and `sm filter-values judge` reported, and + the probe CSVs the execution tier returned. Four grades, and only measurement earns `model_gap`; a + join that could not be graded leaves the fan-out check on its aggregate `unresolved`, said out + loud. `findings` writes a run's findings, the person's own defects listed apart, and every row's + ledger. Three shared references describe the row, the ledger and how a supplied statement is run + the way the AI's own SQL runs. Nothing here runs SQL or writes to the semantic model. (ACE-115) + ### Fixed - **A chain of joins is no longer reported as a chasm trap.** The aggregates section flagged two diff --git a/plugins/agami/scripts/reconcile.py b/plugins/agami/scripts/reconcile.py index dd850dfd..f5a8b176 100644 --- a/plugins/agami/scripts/reconcile.py +++ b/plugins/agami/scripts/reconcile.py @@ -23,6 +23,13 @@ # Band an observed number, ready to paste as a golden item's `bounds`: python3 reconcile.py band --value 47238221 --tolerance 0.01 + + # Read any of the four input shapes a person brings into evidence rows: + # (a) questions, (b) questions with the SQL they trust, (c) labels with numbers, + # (d) labels with numbers and the SQL behind each. Several files merge by label. + # A second file merges by label, so it needs a label column: `label,sql` reads as the SQL + # behind each tile; a bare .sql file has no labels and stands as its own rows. + python3 reconcile.py intake --file tiles.csv --file sql.csv --source "the finance dashboard" """ from __future__ import annotations @@ -270,6 +277,910 @@ def band(value: float, *, tolerance: float = 0.01) -> dict: } +# --- Intake --------------------------------------------------------------- +# +# Any input a person brings reduces to rows of three optional fields: the question, the +# statement, the expected number. The four shapes the skill names are subsets of that row: +# (a) questions only (b) questions with the SQL the person trusts +# (c) labels with numbers (d) labels with numbers and the SQL behind each tile +# `intake` reads all four and says which it saw. The number path is untouched: a two-column +# CSV still goes through `parse_csv`, and a third column that is not SQL is still glued onto +# the label as context, exactly as `parse` always did. + +_STATEMENT_RE = re.compile(r"^\s*(with|select)\b", re.IGNORECASE) + +# Header names a person is likely to type, folded to the field each stands for. A header is +# recognised by NAME rather than by position so a `question,sql` file and a `label,value,sql` file +# both read the way they were written. +_HEADER_FIELDS: dict[str, str] = { + "label": "label", "metric": "label", "tile": "label", "name": "label", "kpi": "label", + "value": "value", "expected": "value", "expected_value": "value", "number": "value", + "amount": "value", "actual": "value", + "sql": "statement", "statement": "statement", "query": "statement", + "question": "question", "prompt": "question", +} + +def _fold(text: str) -> str: + """Case and whitespace fold, the only normalization a label match is allowed.""" + return re.sub(r"\s+", " ", text.strip()).lower() + + +def _is_statement(cell: str | None) -> bool: + return bool(cell) and _STATEMENT_RE.match(cell) is not None + + +def _row_shape(row: dict) -> str: + has_statement = row["statement"] is not None + has_expected = row["expected"] is not None + if has_statement and has_expected: + return "d" + if has_statement: + return "b" + if has_expected: + return "c" + return "a" + + +def _new_row(*, file: str, line: int, source: str | None, label: str | None = None, + question: str | None = None, statement: str | None = None, + raw_value: str | None = None) -> dict: + row = { + "label": label or None, + "question": question or None, + "statement": statement.strip().rstrip(";").strip() if statement else None, + "expected": parse_value(raw_value) if raw_value is not None else None, + "raw_value": raw_value if raw_value not in (None, "") else None, + "provenance": {"shape": None, "source": source, "file": file, "line": line, "graded": None}, + } + row["provenance"]["shape"] = _row_shape(row) + return row + + +def _header_map(first: list[str], rest: list[list[str]]) -> dict[int, str] | None: + """Which field each column holds, when the first row is a header; None when it is data. + + Two ways a row is a header. Every cell names a field this module knows, which is how a + `question,sql` or `label,value,sql` file declares itself. Or, the legacy two-column case + `parse_csv` has always handled: a second cell that is neither a number nor a statement, over a + file whose later rows do carry numbers there. + """ + cells = [c.strip() for c in first] + if cells and all(_fold(c) in _HEADER_FIELDS for c in cells if c): + return {i: _HEADER_FIELDS[_fold(c)] for i, c in enumerate(cells) if c} + if (len(cells) >= 2 and parse_value(cells[1]) is None and not _is_statement(cells[1]) + and any(len(r) >= 2 and parse_value(r[1]) is not None for r in rest)): + fields = {0: "label", 1: "value"} + for i in range(2, len(cells)): + fields[i] = "statement" if _fold(cells[i]) in ("sql", "statement", "query") else "extra" + return fields + return None + + +def _row_from_named(cells: list[str], fields: dict[int, str], *, file: str, line: int, + source: str | None) -> tuple[dict | None, str | None]: + got: dict[str, str] = {} + extras: list[str] = [] + for i, cell in enumerate(cells): + cell = cell.strip() + if not cell: + continue + field = fields.get(i, "extra") + if field == "extra": + extras.append(cell) + elif field == "statement" and not _is_statement(cell): + # A `sql` column holding something that is not a statement is context, not SQL. + extras.append(cell) + else: + got[field] = cell + label = got.get("label") + if label and extras: + label = f"{label} ({', '.join(extras)})" + raw = got.get("value") + if raw is not None and parse_value(raw) is None: + return None, f"the value {raw!r} could not be read as a number" + if not any(k in got for k in ("label", "question", "statement", "value")): + return None, "no question, statement or number in the row" + return _new_row(file=file, line=line, source=source, label=label, + question=got.get("question"), statement=got.get("statement"), + raw_value=raw), None + + +def _row_from_positional(cells: list[str], *, file: str, line: int, + source: str | None) -> tuple[dict | None, str | None]: + """A data row with no header to name its columns, read by shape.""" + cells = [c.strip() for c in cells] + if len(cells) == 1: + text = cells[0] + if _is_statement(text): + return _new_row(file=file, line=line, source=source, statement=text), None + return _new_row(file=file, line=line, source=source, question=text), None + first, second, rest = cells[0], cells[1], cells[2:] + if _is_statement(second): + return _new_row(file=file, line=line, source=source, question=first, statement=second), None + if parse_value(second) is None: + return None, f"the second column {second!r} is neither a number nor a statement" + statement = None + extras = [] + for cell in rest: + if _is_statement(cell) and statement is None: + statement = cell + elif cell: + extras.append(cell) + label = f"{first} ({', '.join(extras)})" if extras else first + return _new_row(file=file, line=line, source=source, label=label, statement=statement, + raw_value=second), None + + +def _rows_from_json(items: Any, *, file: str, source: str | None) -> tuple[list[dict], list[dict]]: + rows: list[dict] = [] + skipped: list[dict] = [] + if not isinstance(items, list): + return rows, [{"file": file, "line": 1, "reason": "a JSON input must be a list"}] + for n, item in enumerate(items, 1): + if isinstance(item, str): + row, why = _row_from_positional([item], file=file, line=n, source=source) + elif isinstance(item, dict): + cells: list[str] = [] + fields: dict[int, str] = {} + for key, value in item.items(): + field = _HEADER_FIELDS.get(_fold(str(key))) + if field is None or value is None: + continue + fields[len(cells)] = field + cells.append(str(value)) + row, why = _row_from_named(cells, fields, file=file, line=n, source=source) + else: + row, why = None, "an item must be a string or an object" + if row is None: + skipped.append({"file": file, "line": n, "reason": why}) + else: + rows.append(row) + return rows, skipped + + +def _rows_from_file(path: Path, source: str | None) -> tuple[list[dict], list[dict]]: + """One file's rows and the lines it could not use. The extension decides how lines are cut: + `.json` is a list, `.sql` is statements split on `;`, `.txt` and `.md` are one question per + line, and everything else is CSV.""" + file = path.name + text = path.read_text(encoding="utf-8") + suffix = path.suffix.lower() + if suffix == ".json": + return _rows_from_json(json.loads(text), file=file, source=source) + if suffix == ".sql": + rows, skipped = [], [] + for n, stmt in enumerate((s for s in text.split(";") if s.strip()), 1): + # The same test a CSV cell gets: anything that is not a SELECT or a WITH is context or + # a mistake, and never reaches the tier as a statement the person supplied. + if _is_statement(stmt): + rows.append(_new_row(file=file, line=n, source=source, statement=stmt.strip())) + else: + skipped.append({"file": file, "line": n, "text": stmt.strip()[:80], + "reason": "not a SELECT or WITH statement"}) + return rows, skipped + if suffix in (".txt", ".md") or ("," not in text and "\t" not in text): + rows = [] + for n, line in enumerate(text.splitlines(), 1): + if line.strip(): + rows.append(_row_from_positional([line], file=file, line=n, source=source)[0]) + return rows, [] + with path.open(newline="", encoding="utf-8") as fh: + numbered = [(n, r) for n, r in enumerate(csv.reader(fh), 1) if r and any(c.strip() for c in r)] + if not numbered: + return [], [] + fields = _header_map(numbered[0][1], [r for _n, r in numbered[1:]]) + data = numbered[1:] if fields is not None else numbered + rows, skipped = [], [] + for n, cells in data: + if fields is not None: + row, why = _row_from_named(cells, fields, file=file, line=n, source=source) + else: + row, why = _row_from_positional(cells, file=file, line=n, source=source) + if row is None: + skipped.append({"file": file, "line": n, "reason": why}) + else: + rows.append(row) + return rows, skipped + + +def _merge_by_label(rows: list[dict]) -> list[dict]: + """A statement whose label matches a tile's label joins that tile's row; anything unmatched + keeps its own row. Matching is the fold only, so `q3 revenue` meets `Q3 Revenue` and nothing + looser does.""" + tiles: dict[str, dict] = {} + for row in rows: + if row["expected"] is not None and row["statement"] is None and row["label"]: + tiles.setdefault(_fold(row["label"]), row) + merged: list[dict] = [] + for row in rows: + key = _fold(row["label"] or row["question"] or "") + if (row["statement"] is not None and row["expected"] is None and key in tiles + and tiles[key]["statement"] is None): + tile = tiles[key] + tile["statement"] = row["statement"] + tile["provenance"]["shape"] = _row_shape(tile) + tile["provenance"]["merged_from"] = {"file": row["provenance"]["file"], + "line": row["provenance"]["line"]} + continue + merged.append(row) + return merged + + +def intake(paths: list[Path], *, source: str | None = None) -> dict: + """Every file's rows, merged by label across files, with the shape that was seen. + + `shape` is one letter when every row has the same shape and `mixed` otherwise; each row also + carries its own under `provenance.shape`, which is what the skill reads row by row. + """ + rows: list[dict] = [] + skipped: list[dict] = [] + for path in paths: + got, missed = _rows_from_file(Path(path).expanduser(), source) + rows.extend(got) + skipped.extend(missed) + rows = _merge_by_label(rows) + shapes = {row["provenance"]["shape"] for row in rows} + shape = next(iter(shapes)) if len(shapes) == 1 else ("mixed" if shapes else None) + return {"shape": shape, "rows": rows, "skipped": skipped} + + +# --- Ledger --------------------------------------------------------------- +# +# One grade per part of a statement the person supplied, read from fixed filenames in the row's +# directory: what happened when it ran (`run.json`), what `sm prepare` and `sm receipt` said about +# it, what `sm join-probes` and `sm filter-values judge` reported, and the probe CSVs the execution +# tier returned. Four grades, and only measurement can earn `model_gap`: +# confirmed the statement and the semantic model agree, and the data backs it +# model_gap the data proves the statement right where the semantic model is missing or wrong +# query_defect the data proves the statement wrong +# unresolved the part could not be checked, and the note says why +# The rules have a dependency in them, and it is applied rather than assumed: a join that could not +# be graded leaves the fan-out check on its aggregate `unresolved`, said out loud, never clean. + +CONFIRMED = "confirmed" +MODEL_GAP = "model_gap" +QUERY_DEFECT = "query_defect" +UNRESOLVED = "unresolved" +# A fifth word that is not a grade: a fact the run states and never judges (rows an inner join +# dropped, a wide column nobody would list). Ranked below `confirmed` so it never decides a row's +# verdict, never blocks an example, and is rendered in its own block. +NOTED = "noted" +_VERDICT_RANK = {QUERY_DEFECT: 3, UNRESOLVED: 2, MODEL_GAP: 1, CONFIRMED: 0, NOTED: -1} + +# Error-classifier kinds that mean the statement itself is wrong, as opposed to the connection. +_STATEMENT_DEFECT_KINDS = {"column_not_found", "table_not_found", "syntax"} +# Guard rules that mean the statement wanted something the semantic model does not expose. +_SCOPE_RULES = {"table_scope", "column_scope"} +# Pre-flight risks that describe how the aggregate itself was written, not how a join fanned it. +_AGGREGATION_RISKS = {"bad_aggregation", "semi_additive"} + + +def _part(part: str, verdict: str, *, kind: str | None = None, depends_on=(), + evidence: dict | None = None, note: str = "") -> dict: + return {"part": part, "verdict": verdict, "kind": kind, "depends_on": list(depends_on), + "evidence": evidence or {}, "note": note} + + +def _load_json(path: Path) -> Any: + """The JSON in `path`; None when the file is absent; `{"error": ...}` when it is empty or is not + JSON. A verb that crashed leaves a zero-byte redirect behind, and that must read as "this input + is unusable", never as "checked and clean".""" + if not path.exists(): + return None + text = path.read_text(encoding="utf-8") + if not text.strip(): + return {"error": "empty_file"} + try: + return json.loads(text) + except json.JSONDecodeError as exc: + return {"error": "unreadable_json", "detail": str(exc).splitlines()[0]} + + +def _usable(payload: Any, key: str) -> "tuple[dict | None, str | None]": + """The payload when it carries `key`, else None and why: absent, empty, an error object from a verb + that exited non-zero, or JSON of another shape.""" + if payload is None: + return None, "was not written" + if not isinstance(payload, dict): + return None, "is not a JSON object" + if payload.get("error"): + return None, f"carries an error ({payload['error']})" + if key not in payload: + return None, f"has no `{key}` key" + return payload, None + + +def _probe_csv(path: Path) -> "list[dict] | str | None": + """A probe's CSV as rows; None when the file is absent; the string `failed` when it is empty. + + The execution tier writes CSV to stdout only on success. A probe that was refused or failed + leaves a zero-byte file behind, and reading that as "the column holds no values" would turn a + failed probe into a definite grade. A header-only file is the legitimately empty result. + """ + if not path.exists(): + return None + if path.stat().st_size == 0: + return "failed" + with path.open(newline="", encoding="utf-8") as fh: + return list(csv.DictReader(fh)) + + +def _first_number(rows, key: str) -> float | None: + """The first row's `key` column as a number. Headers are matched without regard to case, because + one tier upper-cases them; the fall-back to the only column is for a one-column result and never + for a wider one, where it would read the wrong column.""" + if not isinstance(rows, list) or not rows: + return None + row = rows[0] + raw = next((v for k, v in row.items() if k and k.strip().lower() == key.lower()), None) + if raw is None and len(row) == 1: + raw = next(iter(row.values())) + try: + return float(raw) if raw not in (None, "") else None + except (TypeError, ValueError): + return None + + +def _grade_run(run: dict | None) -> list[dict]: + if run is None: + return [_part("runs", UNRESOLVED, note="no run record was found for the statement")] + status, rule, kind = run.get("status"), run.get("rule"), run.get("kind") + if status == "ok": + return [_part("runs", CONFIRMED, note="the statement ran"), + _part("scope", CONFIRMED, note="every table and column it named is in the semantic model")] + if status == "refused": + if rule in _SCOPE_RULES: + return [ + _part("runs", UNRESOLVED, + note=f"the statement was refused before it ran ({rule}); see the scope part"), + _part("scope", MODEL_GAP, kind="scope", + evidence={"rule": rule, "detail": run.get("detail")}, + note="the statement names a table or column the semantic model does not expose"), + ] + if rule == "select_star": + return [_part("runs", QUERY_DEFECT, evidence={"rule": rule}, + note="SELECT * is refused; name the columns")] + return [_part("runs", UNRESOLVED, evidence={"rule": rule}, + note=f"the statement was refused before it ran ({rule})")] + if status == "failed": + if kind in _STATEMENT_DEFECT_KINDS: + return [_part("runs", QUERY_DEFECT, evidence={"kind": kind, "remediation": run.get("remediation")}, + note=f"the database rejected the statement ({kind})")] + return [_part("runs", UNRESOLVED, evidence={"kind": kind}, + note=f"the run failed with {kind}; the statement could not be checked")] + return [_part("runs", UNRESOLVED, note="the statement was not run")] + + +def _join_tables(join: dict) -> tuple[str, str]: + """The two tables a join is between, sorted: from its one written pair when it has one, and + from its endpoint labels otherwise.""" + pairs = join.get("pairs") or [] + if len(pairs) == 1 and len(pairs[0]) == 2: + a, b = pairs[0][0][0], pairs[0][1][0] + else: + a, b = (join.get("endpoints") or ["", ""])[:2] + a, b = _fold(a), _fold(b) + first, second = sorted((a, b)) + return first, second + + +def _join_status(join: dict) -> str: + """The status `sm join-probes` gave the join; one it did not label stays open.""" + return join.get("status") or "undetermined" + + +def _cardinality_result(probes: dict | None, key: str, row_dir: Path) -> dict | None: + """One endpoint's uniqueness: from the semantic model when it declares the column a key, else + from the column's cardinality CSV, which is shared by every join that reads that column.""" + if ((probes or {}).get("unique_by_model") or {}).get(key): + return {"unique": True, "source": "the semantic model declares the column a key"} + got = _probe_csv(row_dir / f"cardinality.{key}.csv") + if not isinstance(got, list) or not got: + return None + total = _first_number(got, "total") + distinct = _first_number(got, "distinct_count") + nulls = _first_number(got, "null_count") or 0.0 + if total is None or distinct is None: + return None + return {"total": total, "distinct": distinct, "nulls": nulls, + "unique": distinct == total - nulls, "source": "probe"} + + +def _grade_joins(probes: dict | None, row_dir: Path) -> list[dict]: + rows: list[dict] = [] + if probes is not None and probes.get("unreadable"): + return [_part("join:*", UNRESOLVED, evidence={"unreadable": probes["unreadable"]}, + note="the join verb could not read the statement, so no join was checked")] + seen: dict[str, int] = {} + for join in (probes or {}).get("joins", []): + a, b = _join_tables(join) + label = f"{a}-{b}" + # Two joins between the same two tables in one statement are two parts, not one: keyed by + # the same label, the second would silently overwrite the first's grade. + seen[label] = seen.get(label, 0) + 1 + if seen[label] > 1: + label = f"{label}#{seen[label]}" + jid = join.get("id", "join") + status = _join_status(join) + planned = join.get("probes") or {} + declared_pairs = join.get("declared_pairs", []) + written = {"pairs": join.get("pairs"), "predicate": join.get("predicate")} + + overlaps: list[float | None] = [] + overlap_failed = False + for i, _probe in enumerate(planned.get("overlap", [])): + got = _probe_csv(row_dir / f"{jid}.overlap.{i}.csv") + if got == "failed": + overlap_failed = True + elif got is not None: + overlaps.append(_first_number(got, "matched")) + card = {key: result for key in planned.get("cardinality", []) + if (result := _cardinality_result(probes, key, row_dir)) is not None} + hits = [m for m in overlaps if m is not None] + any_overlap = any(m > 0 for m in hits) + one_row_on_right = _one_row_on_right(join, probes) + + if status in ("undeclarable", "undetermined"): + rows.append(_part(f"join:{label}", UNRESOLVED, evidence=written, + note=join.get("not_probed_because") + or "the join could not be resolved to two declared tables")) + continue + if status == "declared": + rows.append(_part(f"join:{label}", CONFIRMED, evidence={"declared_pairs": declared_pairs}, + note="the join is on the key the semantic model declares")) + elif status == "wrong_key": + rows.append(_part(f"join:{label}", QUERY_DEFECT, + evidence={"declared_pairs": declared_pairs, **written}, + note="the join is on a different key than the one the semantic model declares")) + elif join.get("too_big_to_probe") or not planned.get("overlap"): + rows.append(_part(f"join:{label}", UNRESOLVED, evidence=written, + note="the join is not declared and no probe could be run: " + + (join.get("not_probed_because") or "no probe was planned"))) + elif any_overlap: + rows.append(_part(f"join:{label}", MODEL_GAP, kind="relationship", + evidence={"overlap": hits, **written}, + note="the join is not declared, and its keys resolve in the data")) + elif hits and not overlap_failed: + rows.append(_part(f"join:{label}", QUERY_DEFECT, evidence={"overlap": hits, **written}, + note="the join is not declared, and its keys never meet in the data")) + else: + # No hit, or a hit beside a probe that failed: half the evidence is not evidence. + rows.append(_part(f"join:{label}", UNRESOLVED, evidence={"overlap": hits, **written}, + note="the join is not declared and " + + ("a probe file is empty, so that probe likely failed; the rest is not enough to decide" + if overlap_failed else "no probe result was supplied"))) + # Whether this join brings in one row at most per row of the table it joins to, by the + # semantic model's own word. The aggregate grader reads it; nothing is re-derived there. + rows[-1]["evidence"]["one_row_on_right"] = one_row_on_right + + # The probe rows: whenever probes were planned or answered. A declared join plans none. + if hits or planned.get("overlap"): + if any_overlap: + rows.append(_part(f"join_key:{label}", CONFIRMED, evidence={"overlap": hits}, + note="sampled keys from one side exist on the other")) + elif hits and not overlap_failed: + rows.append(_part(f"join_key:{label}", QUERY_DEFECT, evidence={"overlap": hits}, + note="no sampled key from either side exists on the other")) + else: + rows.append(_part(f"join_key:{label}", UNRESOLVED, evidence={"overlap": hits}, + note="an overlap probe result is missing or its file is empty")) + if card or planned.get("cardinality"): + uniques = sorted(k for k, v in card.items() if v["unique"]) + if len(card) >= 2 and uniques: + rows.append(_part(f"cardinality:{label}", CONFIRMED, + evidence={"one_side": uniques[0], "sides": card}, + note=f"{uniques[0]} is unique, so the join does not multiply rows")) + elif len(card) >= 2: + rows.append(_part(f"cardinality:{label}", QUERY_DEFECT, evidence={"sides": card}, + note="both sides repeat, so the join multiplies rows")) + else: + rows.append(_part(f"cardinality:{label}", UNRESOLVED, evidence={"sides": card}, + note="no cardinality result for both sides")) + rows.extend(_dropped_rows(join, label, jid, row_dir)) + return rows + + +def _one_row_on_right(join: dict, probes: dict | None) -> bool: + """True when the table this join introduces (its right endpoint) contributes one row at most per + row already there: it is the one side of the declared relationship the statement actually wrote, + or its written column is unique by the semantic model. False for a self-join and for anything + the model did not say. Sound for a chain, because each such join leaves the row count alone.""" + endpoints = join.get("endpoints") or ["", ""] + left_key, right_key = _fold(str(endpoints[0])), _fold(str(endpoints[-1])) + if not right_key or left_key == right_key: + return False + matched_one_sides = {_fold(str(side)) for edge in (join.get("declared_cardinality") or []) + if edge.get("matched") for side in (edge.get("one_side") or [])} + if right_key in matched_one_sides: + return True + pairs = join.get("pairs") or [] + if len(pairs) == 1 and len(pairs[0]) == 2: + # The probe file keys this map with the semantic model's own spelling; the pair carries the + # statement's, lowercased. Folded on both sides, so an uppercase-introspected model + # (`customers.ID`) still says its key is unique. + unique = {_fold(str(k)): v for k, v in ((probes or {}).get("unique_by_model") or {}).items()} + for table, column in pairs[0]: + if _fold(str(table)) == right_key and unique.get(_fold(f"{table}.{column}")): + return True + return False + + +def _dropped_rows(join: dict, label: str, jid: str, row_dir: Path) -> list[dict]: + """The rows an inner join left behind, said and never judged. No probe planned → nothing to say.""" + probe = join.get("dropped_rows_probe") + if not probe: + return [] + left_t, right_t = probe.get("left"), probe.get("right") + unexamined = probe.get("unexamined") + got = _probe_csv(row_dir / f"{jid}.dropped_rows.csv") + # Both numbers by their own header, never the one-column fall-back: a result with one column + # would otherwise read as N of N dropped and be stated as a fact. + total = _named_number(got, "total") + dropped = _named_number(got, "dropped") + if total is None or dropped is None: + return [_part(f"dropped_rows:{label}", NOTED, evidence={"left": left_t, "right": right_t}, + note="the dropped-rows probe was not run or failed; nothing is claimed")] + t, d = int(total), int(dropped) + said = (f"no {left_t} row is dropped by this join" if d == 0 + else f"{d} of {t} {left_t} rows have no {right_t} partner and are dropped by this inner join") + said += "; counted over the whole table, before the statement's own filters" + if unexamined: + said += f"; rows of {unexamined} with no {left_t} partner were not counted" + return [_part(f"dropped_rows:{label}", NOTED, + evidence={"total": t, "dropped": d, "left": left_t, "right": right_t, "unexamined": unexamined}, + note=said)] + + +def _named_number(rows, key: str) -> float | None: + """The first row's `key` column as a number, by header only, whatever the header's case.""" + if not isinstance(rows, list) or not rows: + return None + raw = next((v for k, v in rows[0].items() if k and k.strip().lower() == key.lower()), None) + try: + return float(raw) if raw not in (None, "") else None + except (TypeError, ValueError): + return None + + +def _joins_named(label: str, join_rows: list[dict]) -> list[str]: + """The `join:` parts whose two tables both appear in a pre-flight join label.""" + words = set(re.findall(r"[a-z0-9_]+", _fold(label))) + out = [] + for row in join_rows: + if not row["part"].startswith("join:"): + continue + a, b = row["part"][len("join:"):].split("#", 1)[0].split("-", 1) + if a in words and b in words: + out.append(row["part"]) + return out + + +def _grade_aggregates(prepare: dict | None, join_rows: list[dict], probes: dict | None = None) -> list[dict]: + if prepare is None: + return [] + if prepare.get("unchecked"): + return [_part("fan_out:*", UNRESOLVED, evidence={"unchecked": prepare["unchecked"]}, + note=f"the pre-flight did not run: {prepare['unchecked']}")] + rows: list[dict] = [] + by_part = {row["part"]: row for row in join_rows} + # Every join the statement wrote, not the aggregate's own `joins` list: the pre-flight fills + # that list from multiplying findings only, so it is empty for exactly the aggregate this rule + # is for. The rule needs every written join listed (none dropped at the cap, the verb having + # read the statement), every one confirmed, and every one bringing in one row at most. + join_parts = [row for row in join_rows if row["part"].startswith("join:") and row["part"] != "join:*"] + one_row_joins = ( + isinstance(probes, dict) and probes.get("unreadable") is None + and probes.get("dropped") == 0 and probes.get("joins_written") == len(probes.get("joins") or []) + and bool(join_parts) + and all(row["verdict"] == CONFIRMED and row["evidence"].get("one_row_on_right") for row in join_parts) + ) + for agg in prepare.get("aggregates", []): + text = agg.get("aggregate", "?") + risks = {f.get("risk") for f in agg.get("findings", [])} + deps = sorted({p for label in agg.get("joins", []) for p in _joins_named(label, join_rows)}) + weak = [p for p in deps if by_part[p]["verdict"] != CONFIRMED] + if weak: + rows.append(_part(f"fan_out:{text}", UNRESOLVED, depends_on=deps, + note=f"the join {weak[0][len('join:'):]} this total depends on is " + f"{by_part[weak[0]]['verdict']}, so the fan-out check has no " + "cardinality to reason from")) + elif agg.get("status") == "multiplied": + if risks and risks <= {"fan_out_invariant"}: + rows.append(_part(f"fan_out:{text}", CONFIRMED, depends_on=deps, + note="a join multiplies the rows, but this aggregate cannot move")) + else: + named = sorted(risks - {"fan_out_invariant"}) or ["multiplied"] + rows.append(_part(f"fan_out:{text}", QUERY_DEFECT, depends_on=deps, + evidence={"risks": named, "joins": agg.get("joins", [])}, + note=f"a join multiplies the rows this total is computed from " + f"({', '.join(named)})")) + elif agg.get("status") == "not_multiplied": + rows.append(_part(f"fan_out:{text}", CONFIRMED, depends_on=deps, + note="no join multiplies the rows behind this aggregate")) + elif one_row_joins: + rows.append(_part(f"fan_out:{text}", CONFIRMED, depends_on=[row["part"] for row in join_parts], + evidence={"joins": [row["part"] for row in join_parts]}, + note="every join the statement writes brings in one row at most, so no join " + "multiplies this aggregate")) + else: + # The pre-flight names the blindness it hit; the note repeats it rather than blaming a join. + reason = agg.get("reason") or "no reason was given" + rows.append(_part(f"fan_out:{text}", UNRESOLVED, depends_on=deps, evidence={"reason": agg.get("reason")}, + note=f"the pre-flight could not bind this aggregate to one table: {reason}")) + bad = sorted(risks & _AGGREGATION_RISKS) + if bad: + rows.append(_part(f"aggregation:{text}", QUERY_DEFECT, evidence={"risks": bad}, + note=f"the aggregate is not legal over this column ({', '.join(bad)})")) + else: + rows.append(_part(f"aggregation:{text}", CONFIRMED, note="the aggregate is legal over its column")) + return rows + + +def _grade_filters(receipt: dict | None) -> list[dict]: + rows: list[dict] = [] + for item in ((receipt or {}).get("tables") or {}).get("items", []): + table = item.get("ref") or item.get("qname") or "?" + for flt in item.get("filters", []) or []: + part = f"default_filter:{table}:{flt.get('expr')}" + status = flt.get("status") + if status == "applied": + rows.append(_part(part, CONFIRMED, note="the declared filter is applied")) + elif status == "omitted": + rows.append(_part(part, MODEL_GAP, kind="filter", evidence={"table": table, "expr": flt.get("expr")}, + note="the semantic model declares this filter and the statement does not apply it")) + else: + rows.append(_part(part, UNRESOLVED, note="whether the declared filter is applied could not be read")) + return rows + + +def _grade_metrics(receipt: dict | None, prepare: dict | None) -> list[dict]: + rows: list[dict] = [] + aggregates = [_fold(a.get("aggregate", "")) for a in (prepare or {}).get("aggregates", [])] + only_bare_counts = bool(aggregates) and all(a == "count(*)" for a in aggregates) + for item in ((receipt or {}).get("columns") or {}).get("items", []): + if item.get("kind") != "output": + continue + column = item.get("column", "?") + status = item.get("status") + if status == "matched": + sources = [str(t) for t in (item.get("source_tables") or [])] + read = _tables_read(receipt) + if sources and not (set(sources) & read): + # The receipt matched by shape: the qualifiers were stripped to compare, so a + # `SUM(amount)` on one table equals a metric's `SUM(amount)` on another. + rows.append(_part(f"metric:{column}", UNRESOLVED, + evidence={"metric": item.get("name"), "source_tables": sources, + "tables_read": sorted(read)}, + note=f"matched the metric {item.get('name')!r}, defined on " + f"{', '.join(sources)}, which this statement does not read; " + "the expression matches by shape only")) + else: + rows.append(_part(f"metric:{column}", CONFIRMED, evidence={"metric": item.get("name")}, + note="the output matches a defined metric")) + elif only_bare_counts: + rows.append(_part(f"metric:{column}", CONFIRMED, + note="a bare count matches no metric by design")) + elif status == "unmatched": + rows.append(_part(f"metric:{column}", MODEL_GAP, kind="metric", evidence={"column": column}, + note="the output matches no metric the semantic model defines")) + else: + # `undetermined` is the receipt saying it could not tell (an ambiguous binding, a column + # behind a CTE, a declaration it could not read). A failure to read is never a gap. + rows.append(_part(f"metric:{column}", UNRESOLVED, evidence={"column": column, "status": status}, + note="whether the output matches a metric could not be read")) + return rows + + +def _tables_read(receipt: dict | None) -> set[str]: + """The bare, folded names of every table the statement read, from the receipt's `tables` items.""" + out: set[str] = set() + for item in ((receipt or {}).get("tables") or {}).get("items", []): + for name in (item.get("qname"), item.get("ref")): + if name: + out.add(_fold(str(name)).split(".")[-1]) + return out + + +def _grade_literals(judge: dict | None) -> list[dict]: + rows: list[dict] = [] + if judge is not None and judge.get("unreadable"): + return [_part("literal:*", UNRESOLVED, evidence={"unreadable": judge["unreadable"]}, + note="the filter-values verb could not read the statement, so no value was checked")] + for lit in (judge or {}).get("literals", []): + part = f"literal:{lit.get('table')}.{lit.get('column')}={lit.get('literal')}" + verdict = lit.get("verdict", UNRESOLVED) + rows.append(_part(part, verdict, kind="description" if verdict == MODEL_GAP else None, + evidence={"tier": lit.get("tier"), "op": lit.get("op"), + "near_miss": lit.get("near_miss"), "observed": lit.get("observed"), + "rows_with_value": lit.get("rows_with_value")}, + note=lit.get("note", ""))) + # One part per filtered COLUMN: did the semantic model declare its values at all? Said once, + # not once per literal, and only a column that holds a short list of values is a gap; a wide + # column is noted, because nobody would list it. + for _key, fact in sorted(((judge or {}).get("columns") or {}).items()): + part = f"values_declared:{fact.get('table')}.{fact.get('column')}" + declared, distinct, n = fact.get("declared"), fact.get("distinct"), fact.get("observed_count") + evidence = {"declared": declared, "distinct": distinct, "observed_count": n} + if declared == "populated": + rows.append(_part(part, CONFIRMED, evidence=evidence, note="the semantic model lists this column's values")) + elif fact.get("sensitive"): + rows.append(_part(part, NOTED, evidence=evidence, + note="sensitive column: its values are never listed, so no list is expected")) + elif distinct == "listed": + note = ("introspection found a low-cardinality column and nobody decoded its values" + if declared == "empty" else + f"the column holds {n} distinct values and the semantic model lists none of them") + rows.append(_part(part, MODEL_GAP, kind="description", evidence=evidence, note=note)) + elif distinct == "overflow": + rows.append(_part(part, NOTED, evidence=evidence, + note="the column holds more than 25 distinct values, so no value list is expected")) + elif distinct == "empty": + rows.append(_part(part, NOTED, evidence=evidence, + note="the column returned no values at all, so nothing says whether a list is expected")) + else: + rows.append(_part(part, UNRESOLVED, evidence=evidence, + note="whether the semantic model should list this column's values could not be " + f"checked: the distinct probe was {distinct or 'not run'}")) + return rows + + +def _grade_claims(claims: dict | None) -> list[dict]: + rows: list[dict] = [] + wanted = {"filter_predicates": "predicates", "date_window": "date_window"} + for claim in (claims or {}).get("claims", []): + part = wanted.get(claim.get("name")) + if part is None: + continue + evidence = {"status": claim.get("status"), "generated": claim.get("generated"), + "golden": claim.get("golden")} + if claim.get("status") == "agrees": + rows.append(_part(part, CONFIRMED, evidence=evidence, note="both statements agree")) + elif claim.get("status") == "differs": + rows.append(_part(part, UNRESOLVED, evidence=evidence, + note="the two statements differ here; which is right is not decided by this comparison")) + else: + rows.append(_part(part, UNRESOLVED, evidence=evidence, + note="this claim could not be read on one side")) + return rows + + +def ledger(row_dir: Path, *, with_claims: bool = False) -> dict: + """Every part of the statement in `row_dir`, graded, and the verdict the weakest part decides.""" + row_dir = Path(row_dir) + run = _load_json(row_dir / "run.json") + prepare = _load_json(row_dir / "statement-prepare.json") + receipt = _load_json(row_dir / "statement-receipt.json") + probes = _load_json(row_dir / "join-probes.json") + judge = _load_json(row_dir / "filter-values.judge.json") + claims = _load_json(row_dir / "claims.json") if with_claims else None + + rows = _grade_run(run) + # After a run that succeeded, every input the later steps write is expected. One that is absent, + # empty, or an error object is a part of the statement that was NOT checked, said as such: the + # alternative, grading only what is there, makes a crashed verb read as a clean statement. + ran = isinstance(run, dict) and run.get("status") == "ok" + expected = (("statement-prepare.json", prepare, "aggregates", "fan_out:*"), + ("statement-receipt.json", receipt, "tables", "receipt:*"), + ("join-probes.json", probes, "joins", "join:*"), + ("filter-values.judge.json", judge, "literals", "literal:*")) + checked: dict[str, dict | None] = {} + for name, payload, key, part in expected: + got, why = _usable(payload, key) + checked[name] = got + if ran and got is None: + rows.append(_part(part, UNRESOLVED, evidence={"file": name, "problem": why}, + note=f"{name} {why}, so this part of the statement was not checked")) + prepare, receipt = checked["statement-prepare.json"], checked["statement-receipt.json"] + probes, judge = checked["join-probes.json"], checked["filter-values.judge.json"] + if with_claims: + claims, _why = _usable(claims, "claims") + join_rows = _grade_joins(probes, row_dir) + rows.extend(join_rows) + rows.extend(_grade_aggregates(prepare, join_rows, probes)) + rows.extend(_grade_filters(receipt)) + rows.extend(_grade_metrics(receipt, prepare)) + rows.extend(_grade_literals(judge)) + rows.extend(_grade_claims(claims)) + + counts = {v: 0 for v in _VERDICT_RANK} + for row in rows: + counts[row["verdict"]] = counts.get(row["verdict"], 0) + 1 + verdict = max((row["verdict"] for row in rows), key=lambda v: _VERDICT_RANK.get(v, 2)) + return {"rows": rows, "verdict": verdict, "counts": counts} + + +# --- Findings ------------------------------------------------------------- + + +# A table or alias qualifier in front of a column name, for folding `o.status` and `orders.status`. +_QUALIFIER_RE = re.compile(r"\b[a-z_][a-z0-9_]*\.(?=[a-z_])") + + +def _finding_key(row: dict) -> str | None: + """One key per problem, so the same gap seen from two statements counts once.""" + part, kind = row["part"], row.get("kind") + if kind == "relationship" and part.startswith("join:"): + return f"relationship:{part[len('join:'):]}" + if kind == "filter" and part.startswith("default_filter:"): + table, _sep, expr = part[len("default_filter:"):].partition(":") + # The receipt spells the filter with the statement's own alias (`o.status`), so the same + # declared filter seen through two aliases would be two findings. The qualifier is dropped. + return f"filter:{table}:{_QUALIFIER_RE.sub('', _fold(expr))}" + if kind == "metric" and part.startswith("metric:"): + return f"metric:{_fold(part[len('metric:'):])}" + if kind == "scope": + return f"scope:{row.get('evidence', {}).get('rule') or 'scope'}" + if kind == "description" and part.startswith("literal:"): + return f"description:{_fold(part[len('literal:'):].split('=', 1)[0])}" + if kind == "description" and part.startswith("values_declared:"): + # The same family as a stale list: one column whose declared values are missing or wrong. + return f"description:{_fold(part[len('values_declared:'):])}" + return f"{kind or 'other'}:{_fold(part)}" + + +def findings(run_dir: Path) -> dict: + """The run's findings, its defects, and every row's ledger, written beside `rows.jsonl`. + + A finding is one place the semantic model was shown to be missing or wrong, with every row that + showed it. A defect is one part of a person's statement the data proved wrong, listed apart so + nothing about the semantic model is proposed from it. A statement the AI got wrong while every + part of the person's statement held is a finding of kind `example`: the fix is a worked example, + not a change to a definition. + """ + run_dir = Path(run_dir) + records: list[dict] = [] + rows_path = run_dir / "rows.jsonl" + if rows_path.exists(): + for n, line in enumerate(rows_path.read_text(encoding="utf-8").splitlines(), 1): + if line.strip(): + record = json.loads(line) + record.setdefault("row", n) + records.append(record) + ledgers: dict[str, dict] = {} + grouped: dict[str, dict] = {} + defects: list[dict] = [] + for record in records: + n = record["row"] + row_dir = run_dir / "rows" / str(n) + graded = ledger(row_dir, with_claims=True) if row_dir.exists() else None + if graded is not None: + ledgers[str(n)] = graded + parts = graded["rows"] if graded else [] + evidence_base = {"row": n, "question": record.get("question"), + "statement": record.get("statement"), "expected": record.get("expected"), + "claims": record.get("claims")} + for part in parts: + if part["verdict"] == QUERY_DEFECT: + defects.append({"row": n, "part": part["part"], "note": part["note"]}) + elif part["verdict"] == MODEL_GAP: + key = _finding_key(part) + entry = grouped.setdefault(key, {"key": key, "kind": part.get("kind"), "evidence": []}) + entry["evidence"].append({**evidence_base, "part": part["part"], + "note": part["note"], "ledger": part["evidence"]}) + if record.get("words"): + entry["words"] = record["words"] + # An example is offered only when the person's statement held on EVERY part. A part left + # open, or a row that was never graded at all, is not a statement that held. + clean = bool(parts) and all(p["verdict"] in (CONFIRMED, NOTED) for p in parts) + if record.get("status") == "mismatch" and clean and record.get("question"): + key = f"example:{_fold(record['question'])}" + entry = grouped.setdefault(key, {"key": key, "kind": "example", "evidence": []}) + entry["evidence"].append({**evidence_base, "part": None, + "note": "the statement held on every part and the AI's answer differed", + "ledger": {}}) + result = { + "findings": sorted(grouped.values(), key=lambda f: f["key"]), + "query_defects": sorted(defects, key=lambda d: (d["row"], d["part"])), + "ledger": ledgers, + } + for entry in result["findings"]: + entry["evidence"].sort(key=lambda e: e["row"]) + (run_dir / "findings.json").write_text(json.dumps({"findings": result["findings"]}, indent=2), encoding="utf-8") + (run_dir / "query_defects.json").write_text(json.dumps(result["query_defects"], indent=2), encoding="utf-8") + (run_dir / "ledger.json").write_text(json.dumps(ledgers, indent=2), encoding="utf-8") + return result + + # --- CLI ------------------------------------------------------------------ @@ -294,8 +1205,63 @@ def main(argv: list[str] | None = None) -> int: p_band.add_argument("--value", required=True) p_band.add_argument("--tolerance", default="0.01") + p_intake = sub.add_parser("intake", help="Read any input shape a person brings into evidence rows.") + p_intake.add_argument("--file", action="append", required=True, dest="files") + p_intake.add_argument("--source", default=None, help="the person's own words for where this came from") + + p_ledger = sub.add_parser("ledger", help="Grade every part of a supplied statement from the files in its row directory.") + p_ledger.add_argument("--row-dir", required=True, dest="row_dir") + p_ledger.add_argument("--with-claims", action="store_true", dest="with_claims", + help="also read claims.json, the diff against the AI's own statement") + + p_findings = sub.add_parser("findings", help="Write a run's findings, defects and ledgers beside its rows.jsonl.") + p_findings.add_argument("--run-dir", required=True, dest="run_dir") + args = p.parse_args(argv) + if args.cmd == "ledger": + row_dir = Path(args.row_dir).expanduser() + if not row_dir.is_dir(): + print(f"reconcile ledger: row directory not found: {row_dir}", file=sys.stderr) + return 2 + result = ledger(row_dir, with_claims=args.with_claims) + (row_dir / "ledger.json").write_text(json.dumps(result, indent=2), encoding="utf-8") + print(json.dumps(result, indent=2)) + return 0 + + if args.cmd == "findings": + run_dir = Path(args.run_dir).expanduser() + if not run_dir.is_dir(): + print(f"reconcile findings: run directory not found: {run_dir}", file=sys.stderr) + return 2 + if not (run_dir / "rows.jsonl").exists(): + print("reconcile findings: no rows to read; the run wrote no rows.jsonl", file=sys.stderr) + return 4 + print(json.dumps(findings(run_dir), indent=2)) + return 0 + + if args.cmd == "intake": + paths = [Path(f).expanduser() for f in args.files] + missing = [str(p) for p in paths if not p.exists()] + if missing: + print(f"reconcile intake: file not found: {', '.join(missing)}", file=sys.stderr) + return 2 + try: + result = intake(paths, source=args.source) + except (json.JSONDecodeError, UnicodeDecodeError) as exc: + print(f"reconcile intake: could not read the input: {exc}", file=sys.stderr) + return 2 + if not result["rows"]: + # Exit 4, "nothing to do", kept apart from 2 so the skill can say which happened: + # a file it could not open, or a file with no question, statement or number in it. + # The skipped lines and their reasons still go to stdout, so the person hears why. + print(json.dumps(result, indent=2)) + print("reconcile intake: nothing usable in the input; no question, statement or number " + "was found", file=sys.stderr) + return 4 + print(json.dumps(result, indent=2)) + return 0 + if args.cmd == "parse": rows = parse_csv(args.csv) print(json.dumps(rows, indent=2)) diff --git a/plugins/agami/shared/evidence-row.md b/plugins/agami/shared/evidence-row.md new file mode 100644 index 00000000..9b71b61b --- /dev/null +++ b/plugins/agami/shared/evidence-row.md @@ -0,0 +1,99 @@ +# The evidence row + +Shared by `agami-reconcile` (which reads every input through it) and `agami-save-correction` (which +reads a pasted statement the same way). Read this before writing code or prose that consumes what a +person hands over: a screenshot, a spreadsheet, a query they trust, a list of questions, or a mix. + +**The person's query is evidence, never the answer.** Every row here is something to check. Nothing in +a row is trusted until the checks in [`part-ledger.md`](part-ledger.md) have run. + +## The four shapes, and the one row they all become + +Any input reduces to rows of three optional fields plus where it came from: + +```json +{"label": "Q3 Revenue", "question": null, "statement": "SELECT SUM(total) FROM orders WHERE q = 3", + "expected": 4200000.0, "raw_value": "$4.2M", + "provenance": {"shape": "d", "source": "the finance dashboard", "file": "tiles.csv", "line": 2, + "graded": null}} +``` + +| Shape | What the person handed over | `question` | `statement` | `expected` | +|---|---|---|---|---| +| **a** | a list of questions | given | none | none, until the person grades the AI's answer | +| **b** | questions with the SQL they trust | given, or derived from the statement and confirmed | given | produced by running the statement | +| **c** | a dashboard screenshot, or a label-and-number table | a label, turned into a question and confirmed | none | read by code, confirmed by the person | +| **d** | a screenshot or table with the SQL behind each tile | a label | given | read by code, and checked against the statement's own result | + +`provenance.shape` is per row. The input's own `shape` is one letter when every row agrees and +`mixed` otherwise. + +## How `reconcile.py intake` reads an input + +Run it once per input, with every file the person gave: + +```bash +python3 "$AGAMI_PLUGIN_ROOT/scripts/reconcile.py" intake --file [--file ...] \ + [--source ""] > /tmp/agami-reconcile-rows-.json +``` + +Detection is by content, never by asking: + +- **A cell that starts with `SELECT` or `WITH`** is a statement. A `.sql` file is statements split on + `;`, and every chunk gets the same test: one that is not a `SELECT` or a `WITH` is skipped with a + reason, never a statement row. +- **A `.txt` or `.md` file, or a file with no commas or tabs,** is one question per line; a line that + starts with `SELECT` or `WITH` is a statement. +- **A `.json` file** is a list of strings (questions) or objects with any of the keys `label`, + `question`, `value`, `sql`, or their common spellings (`metric`, `tile`, `expected`, `statement`, + `query`, `prompt`, and the like). +- **Anything else is CSV.** A header row is recognised by name (`label`, `metric`, `value`, + `expected`, `sql`, `statement`, `question`, and their common spellings), or by the shape + `parse_csv` always recognised: a non-numeric second cell over rows whose second cells are numbers. + Without a header: one cell is a question; two cells are a question and a statement when the second + is SQL, or a label and a number when it parses; three or more cells are a label, a number, and a + statement when one of the rest is SQL. A third cell that is not SQL is glued onto the label as + context, exactly as `parse` has always done. +- **Numbers are parsed by code**, through `parse_value`, never by the AI eyeballing them. The + vision branch of the skill writes what it read to a CSV first, so it goes through the same reader. + +**Mixed input merges by label.** A statement whose label matches a tile's label, under a case and +whitespace fold, joins that tile's row and the row's shape becomes `d`. An unmatched statement gets +its own row (shape `b`); an unmatched tile stays a number-only row (shape `c`). + +**What is skipped, and what is refused.** A row whose second column is neither a number nor a +statement is skipped with a reason, in `skipped`. A file that does not exist exits `2`. An input with +no question, statement or number anywhere exits `4`. Nothing is guessed. + +## The row record the skill keeps + +Phase 2d of `agami-reconcile` writes one record per row to `rows.jsonl`. Every key that record has +always had stays, with the same meaning: `label`, `question`, `expected`, `actual`, `delta_pct`, +`match`, `status`, `report_path`, `sql`, `recorded`, `error`. These keys are appended after `error`: + +| Key | Holds | +|---|---| +| `provenance` | the block above: `shape`, `source`, `file`, `line`, `graded` (`null`, `right`, `wrong` or `unsure` once the person has graded a shape-a row), and `merged_from` (the file and line of a statement that joined a tile's row by label) | +| `statement` | the person's SQL, verbatim; `null` when they gave none | +| `statement_recorded` | what the person's SQL returned: one cell as `{"columns": [...], "rows": [[v]]}`, or `{"columns": [...], "row_count": n}` for a table. Never the rows of a table | +| `statement_receipt_path` | the receipt of the person's SQL, in the row directory | +| `receipt_path` | the receipt of the AI's SQL | +| `ledger`, `ledger_verdict` | the graded parts and the weakest grade, from `reconcile.py ledger` | +| `comparison` | `{"scalar": }` or `{"result_set": }` | +| `claims` | the `sm claims` diff between the two statements, when both exist | +| `finding_keys` | the keys of the findings this row contributed to | +| `words` | what the person wrote beside a `wrong` grade when they gave no SQL; `null` otherwise | + +## The status a row gets + +`status` keeps its three old values and gains two: + +| Status | When | +|---|---| +| `match` | the numbers agree within tolerance, and every graded part is `confirmed` (or there was no statement to grade) | +| `match_unverified` | the numbers agree, but a part of the person's statement is not `confirmed`. Never offered in Phase 3e: a match nobody could verify may be luck | +| `mismatch` | the numbers differ and the person's statement has no `query_defect`, so the AI is the likelier culprit | +| `expected_doubtful` | the numbers differ and the person's statement has a `query_defect`, so the expected value itself is in doubt. Kept out of the mismatch tally | +| `error` | the row could not run; `sql` and `recorded` are `null`, as they always were | + +`match` (the boolean) stays `reconcile.diff`'s verdict on the numbers alone. diff --git a/plugins/agami/shared/file-layout.md b/plugins/agami/shared/file-layout.md index 2a208341..674a36e6 100644 --- a/plugins/agami/shared/file-layout.md +++ b/plugins/agami/shared/file-layout.md @@ -18,6 +18,7 @@ Everything in here is either a secret, an auth file derived from a secret, or pe | `charts//.html` | Per-query HTML reports | | `exports//.csv` | Per-query CSV exports | | `{review,model,examples-validation,eval}//.html` | Per-profile dashboards. The `eval` kind also holds a JSON run artifact per run — the answer key and the generated statement, side by side | +| `reconcile//` | One reconcile run: `rows.jsonl`, `ledger.json`, `findings.json`, `query_defects.json`, and `rows//` holding each supplied statement, its run record, receipts, probe plans and probe CSVs. Question text and SQL, never result rows beyond one recorded cell. See [`part-ledger.md`](part-ledger.md) | | `serve/`, `tunnels/` | The copied MCP server; SSH tunnel scripts | | `.duckdb_init_*.sql` | Ephemeral, chmod-600 — federation init files, deleted after the query | diff --git a/plugins/agami/shared/part-ledger.md b/plugins/agami/shared/part-ledger.md new file mode 100644 index 00000000..0779ca46 --- /dev/null +++ b/plugins/agami/shared/part-ledger.md @@ -0,0 +1,105 @@ +# The part ledger + +How a statement a person supplied is graded, one part at a time. Shared by `agami-reconcile` +(Phase 1.5) and `agami-save-correction` (Phase 1d, for a pasted statement). The person's query is +evidence, never the answer: a part is graded against the semantic model and the warehouse, and the +query saying something is never proof of it. + +## Four grades and a note, and only measurement earns a `model_gap` + +| Grade | Means | What follows | +|---|---|---| +| `confirmed` | the statement and the semantic model agree on this part, and the data backs it | nothing | +| `model_gap` | the data proves the statement right where the semantic model is missing it or has it wrong | a finding, for a person to act on | +| `query_defect` | the data proves the statement wrong on this part | reported to the person; nothing about the semantic model changes | +| `unresolved` | the part could not be checked, and the note says why | nothing is written; a row with one, or a row never graded at all, is never kept as an example | +| `noted` | not a grade: a fact the run states and never judges (rows an inner join dropped, a wide column nobody would list) | shown in its own block; never decides the row's verdict and never blocks an example | + +**A check that could not run is never a pass.** And a failed measurement upstream never lends a grade +downstream: a join that could not be graded leaves the fan-out check on its aggregate `unresolved`. + +## The parts, and what grades each one + +| Part id | Read from | Rule | +|---|---|---| +| `runs` | `run.json` | `ok` → confirmed. Failed with `column_not_found`, `table_not_found` or `syntax` → query_defect. Refused for `select_star` → query_defect. Refused for `table_scope` or `column_scope` → unresolved here, and see `scope`. Any other failure → unresolved. No record → unresolved. The evidence carries the classifier's `kind` and `remediation`, never the engine's error text | +| `scope` | `run.json` | a scope refusal → model_gap of kind `scope`: the statement names a table or column the semantic model does not expose. A clean run → confirmed | +| `join:-` | `join-probes.json` + overlap CSVs | the verb's `status` decides: `declared` → confirmed; `wrong_key` (a different key between two tables with a declared relationship) → query_defect, whatever the probe says; `undeclarable` or `undetermined` (a CTE or derived table, a `USING`, a comma join, a declared `on:` nobody could read) → unresolved, with the verb's reason. `undeclared`: keys overlap → model_gap of kind `relationship`; keys never meet → query_defect; not probed, or one overlap probe's file empty → unresolved. A second join between the same two tables is its own part, `join:-#2` | +| `join_key:-` | overlap CSVs | sampled keys from one side exist on the other → confirmed; none do, every probe having answered → query_defect; a result missing or a probe file empty → unresolved. Emitted whenever overlap probes were planned or answered, which a declared join never plans | +| `cardinality:-` | `cardinality...csv`, or the semantic model | one side unique (a declared key, or distinct = total − nulls) → confirmed, naming the side; both sides repeat → query_defect, the join multiplies rows; a side missing → unresolved | +| `fan_out:` | `statement-prepare.json` | a `join:` part it depends on is not confirmed → unresolved. `multiplied` with only `fan_out_invariant` → confirmed. `multiplied` otherwise → query_defect, naming the risk. `not_multiplied` → confirmed. `undetermined` with every written join listed, confirmed, and bringing in one row at most (its right endpoint the one side of the declared relationship the statement wrote, or its written column unique by the semantic model) → confirmed, "every join the statement writes brings in one row at most"; `undetermined` otherwise → unresolved, the note repeating the pre-flight's `reason` (the aggregate names no column; a column attributed to no single table; a name bound to a computed relation; a table the semantic model does not declare). Pre-flight `unchecked` → one `fan_out:*` row, unresolved | +| `aggregation:` | `statement-prepare.json` | a `bad_aggregation` or `semi_additive` risk → query_defect; else confirmed | +| `default_filter:
:` | `statement-receipt.json` `tables.items[].filters` | `applied` → confirmed; `omitted` → model_gap of kind `filter`; `undetermined` → unresolved | +| `metric:` | `statement-receipt.json` `columns.items[]` | `matched` → confirmed; `unmatched` → model_gap of kind `metric`, except when every aggregate in the statement is a bare `count(*)`, which matches no metric by design; `undetermined` (the receipt could not tell) → unresolved, because a failure to read is never a gap; `matched` to a metric whose `source_tables` the statement never reads → unresolved, the match being by shape alone | +| `literal:.=` | `filter-values.judge.json` | the judge's grade, as it stands; a `model_gap` is of kind `description`, the column's list of values being stale. Every grade carries `declared` (`populated`, `empty`, `absent`), and a grade the warehouse decided over an undeclared column says so in its note | +| `values_declared:.` | `filter-values.judge.json` `columns` | one per filtered column. `populated` → confirmed; `absent` or `empty` with the distinct probe `listed` (under 26 values) → model_gap of kind `description`, the same finding family as a stale list; `overflow` → noted, no list is expected of a wide column; `empty` → noted; `failed` or `not_run` → unresolved; a sensitive column → noted | +| `dropped_rows:-` | `.dropped_rows.csv` | noted, never a grade: ` of rows have no partner`, counted over the whole table before the statement's own filters, and naming the other side as not counted (an inner join drops from both); a probe planned but not run → noted, nothing claimed; no probe planned → no part | +| `predicates`, `date_window` | `claims.json`, only with `--with-claims` | `agrees` → confirmed; `differs` or `unknown` → unresolved, with both sides named. A difference is reported, never judged here | + +**The verdict is the weakest part:** `query_defect` outranks `unresolved`, which outranks +`model_gap`, which outranks `confirmed`. The counts travel with it so a reader sees what else was there. + +**An input that is not there is a part that was not checked.** After a run whose `run.json` says +`ok`, the ledger expects `statement-prepare.json`, `statement-receipt.json`, `join-probes.json` and +`filter-values.judge.json`. One that is absent, zero bytes, one JSON error line from a verb that +exited non-zero, or JSON of another shape becomes one open part, `fan_out:*`, `receipt:*`, `join:*` or +`literal:*`, whose evidence names the file and the problem. A verb that could not read the statement +(`unreadable` set) opens `join:*` or `literal:*` the same way. Grading only what happened to be there +would make a crashed verb read as a clean statement. + +## The row directory + +The skill writes one directory per row, `/local/reconcile//rows//`, with fixed +filenames, and `reconcile.py ledger --row-dir [--with-claims]` reads them. The verb is idempotent +and writes `ledger.json` beside the inputs. + +| File | Written by | Holds | +|---|---|---| +| `statement.sql` | the skill | the person's statement, verbatim | +| `run.json` | the skill | `{"status": "ok" \| "refused" \| "failed" \| "not_run", "rule": ..., "kind": ..., "detail": ...}` from the tier's exit, the refusal line, or the error classifier | +| `statement.csv` | the tier | the statement's result; only its shape and one cell are ever copied onward | +| `statement-prepare.json` | `sm prepare --sql-file` | aggregates, findings, `unchecked` | +| `statement-receipt.json` | `sm receipt --sql-file` | the receipt of the person's statement | +| `join-probes.json` | `sm join-probes --sql-file` | every join written with its status, what the semantic model declares about it (`declared_cardinality`, `unique_by_model`), the probes to run, and a top-level `cardinality` map of one probe per column | +| `.overlap..csv` | the tier | the `i`-th overlap probe's `matched` count | +| `.dropped_rows.csv` | the tier | `total, dropped` for the join's left table: its rows with no partner on the right | +| `cardinality.
..csv` | the tier | `total, distinct_count, null_count` for one column, shared by every join that reads it; not written for a column the semantic model declares a key | +| `filter-values.plan.json` | `sm filter-values plan --sql-file` | every typed value and its probes, and a `columns` map of one distinct-values probe per column | +| `.distinct.csv` | the tier | one column's distinct values, bounded one past the enum ceiling | +| `.exists.csv`, `.exists_folded.csv` | the tier | one value's row count, and the folded near miss, run only when `exists` returned 0 | +| `filter-values.judge.json` | `sm filter-values judge` | one grade per typed value | +| `claims.json` | `sm claims` | the diff against the AI's own statement, once both exist | +| `receipt.json` | `sm receipt` | the receipt of the AI's statement | +| `ledger.json` | `reconcile.py ledger` | the graded parts | + +**A zero-byte probe CSV is a probe that failed**, because the tier writes CSV to stdout only on +success. The ledger reads it as unresolved and says so. A header-only CSV is a probe that ran and +found nothing. + +## The findings file + +`reconcile.py findings --run-dir ` reads `rows.jsonl` and every row directory, and writes three +files at the run's root: `ledger.json` (every row's ledger, by row number), `findings.json`, and +`query_defects.json`. + +A finding is one place the semantic model was shown to be missing or wrong, with every row that +showed it: + +```json +{"key": "relationship:customers-orders", "kind": "relationship", + "evidence": [{"row": 1, "part": "join:customers-orders", "question": "...", "statement": "...", + "expected": 4200000.0, "note": "...", "ledger": {...}, "claims": null}], + "words": "what the person wrote, when a grade came with a note"} +``` + +Kinds: `relationship`, `filter`, `metric`, `scope`, `description`, `example`. Keys sort table pairs +and fold expressions to lowercase single-spaced text, with the kind as prefix, so the same missing +join seen from two statements in either order is one finding and a filter gap never collides with a +metric gap. A row whose status is `mismatch` while every part of the person's statement is `confirmed` is a +finding of kind `example`: the AI answered differently from a statement that checks out, and the fix +is a worked example rather than a change to a definition. A part left `unresolved`, or a row that was +never graded, is not a statement that held, and makes no example. Filter keys drop the statement's +alias, so `o.status` and `orders.status` over one declared filter are one finding. + +`query_defects.json` lists `{row, part, note}` for every `query_defect`, apart from the findings, so +nothing about the semantic model is ever proposed from a part the data proved wrong. diff --git a/plugins/agami/shared/statement-check.md b/plugins/agami/shared/statement-check.md new file mode 100644 index 00000000..bc93bd4f --- /dev/null +++ b/plugins/agami/shared/statement-check.md @@ -0,0 +1,69 @@ +# Running a statement a person supplied + +The person's statement takes the road the AI's SQL takes. There is no second road. `agami-query` +Phase 1e says how the profile's tier is invoked, Phase 3a says the two steps every statement goes +through, Phase 3b says what an error means, and Phase 4e.iii.5 says how a receipt is assembled. This +page only says what to do at each step with a statement you did not write, and which file to write it +to. The files are the ones [`part-ledger.md`](part-ledger.md) reads. + +Work in the row's directory, `/local/reconcile//rows//`. Write +`statement.sql` first, verbatim. + +1. **Read-only first.** One `SELECT` or `WITH ... SELECT` per + [`sql-generation-rules.md`](sql-generation-rules.md). Anything else is refused here: write + `run.json` with `status: "not_run"`, mark the row `error`, and probe nothing. +2. **Does it run at all?** Wrap it so it returns no rows, the way seed validation does: + `SELECT 1 FROM () AS _agami_check WHERE 1=0`, through steps 3 and 4. A failure whose + classifier kind is `column_not_found`, `table_not_found` or `syntax` is the person's defect: write + `run.json` with `status: "failed"` and the `kind`, and stop probing. +3. **`sm prepare` on every tier.** `bash "$AGAMI_PLUGIN_ROOT/scripts/sm" prepare "$ROOT" --area + --sql-file statement.sql > statement-prepare.json`. Keep `aggregates`, `findings` and `unchecked`. + It describes and never refuses. +4. **The tier's own tool**, exactly as `agami-query` Phase 1e tabulates it for this profile: psql, + mysql, snowsql, sqlite3, DuckDB, or `"$PY" -m execute_sql --profile --area + --sql-file statement.sql`. **Never `--no-safety`.** Always pass the statement **by file** (the + tool's `-f` or `--sql-file` form), never inline in a shell string: a literal such as `'$(id)'` is + legal SQL and the shell would expand it. stdout goes to `statement.csv`. `run.json` records + `status` (`ok`, `failed`, `refused`, `not_run`), `exit`, the classifier's `kind`, the guard's + `rule`, and its `remediation`; **never the raw stderr**, which can carry the statement and the + engine's error text. +5. **A refusal is a finding, not a crash.** `execute_sql` exits `1` with one JSON line on stderr, + `{"refusal": {"reason", "rule", "detail", "remediation"}}`. Write `run.json` with + `status: "refused"` and that `rule`. `table_scope` and `column_scope` become a `scope: model_gap` + in the ledger: the person wanted a table or column the semantic model does not expose. + `select_star` becomes `runs: query_defect`. Never rewrite the statement and never retry: a + regenerated statement is one the person never wrote. +6. **Other failures** go through [`db_error_classifier.md`](db_error_classifier.md). `auth`, `dsn`, + `network` and `permission` stop the whole run, as Phase 3b stops it; write the `kind` and its + `remediation` and move on. +7. **`sm receipt`** whenever the statement parsed: `bash "$AGAMI_PLUGIN_ROOT/scripts/sm" receipt + "$ROOT" --sql-file statement.sql > statement-receipt.json`. +8. **Probes** go through step 4 only, each written to its own `.sql` file first and passed by path, + each result to its own CSV named as `part-ledger.md` lists: + `bash "$AGAMI_PLUGIN_ROOT/scripts/sm" join-probes "$ROOT" --sql-file statement.sql > + join-probes.json`, then each join's `probes.overlap[i].sql` to `.overlap..csv` and + each entry of the top-level `cardinality` map to `cardinality.
..csv` (skip a null + entry: the semantic model already says that column is unique), and each join's `dropped_rows_probe.sql` to + `.dropped_rows.csv` (skip a null probe); `bash "$AGAMI_PLUGIN_ROOT/scripts/sm" + filter-values plan "$ROOT" --sql-file statement.sql > filter-values.plan.json`, then each + `columns[].distinct` to `.distinct.csv`, each literal's `probes.exists` to + `.exists.csv`, and `probes.exists_folded` to `.exists_folded.csv` only when + `exists` returned 0; then `bash "$AGAMI_PLUGIN_ROOT/scripts/sm" filter-values judge "$ROOT" --plan + filter-values.plan.json --results . > filter-values.judge.json`. + A probe the tier refuses or fails leaves an empty CSV; leave it, the ledger reads it as a probe + that failed. Beside every probe's CSV write `.run.json` with the same fields as step + 4's `run.json`, refusal `rule` included: that file is the record of the probe having run or having + been refused. +9. **Nothing in these steps writes `query_log.jsonl`, and nothing here runs unrecorded.** + `agami-save-correction` reads that log's last successful line as the question to correct, and a + probe there would be corrected instead of the answer. The record of this phase is the row + directory itself: `run.json` for the statement and `.run.json` for every probe, each with + its exit, rule and kind, so every execution and every refusal in this phase is written down. The + AI's own run logs as `agami-query` Phase 5 always has. + +Then `python3 "$AGAMI_PLUGIN_ROOT/scripts/reconcile.py" ledger --row-dir .` grades what was found, +and again with `--with-claims` once `sm claims` has compared the two statements. + +**The person's statement is never run with weaker guards than the AI's.** Every gate that refuses a +generated statement refuses a supplied one, and the refusal is written down as a grade rather than +treated as the run breaking. diff --git a/tests/test_reconcile_intake.py b/tests/test_reconcile_intake.py new file mode 100644 index 00000000..e82a00bd --- /dev/null +++ b/tests/test_reconcile_intake.py @@ -0,0 +1,187 @@ +"""`reconcile.py intake` — any of the four input shapes a person brings, read into evidence rows. + +The skill used to read one shape: a label and a number. A third CSV column was glued onto the label +as text, so a spreadsheet of tile, number and the SQL behind the tile arrived as a mangled label; a +list of questions, or SQL pasted on its own, was not recognised at all. `intake` reads all four and +says which it saw. The number path is unchanged: `parse`, `diff` and `band` are pinned elsewhere, and +a two-column CSV still goes through `parse_csv`. + +Shapes, as the contract names them: (a) questions only, (b) questions with the SQL the person +expects, (c) labels with numbers, (d) labels with numbers and the SQL behind each. Synthetic fixtures +throughout. +""" + +from __future__ import annotations + +import json +import sys +from pathlib import Path + +REPO_ROOT = Path(__file__).resolve().parent.parent +sys.path.insert(0, str(REPO_ROOT / "plugins" / "agami" / "scripts")) + +import reconcile # noqa: E402 +from reconcile import intake # noqa: E402 + + +def _file(tmp_path: Path, name: str, text: str) -> Path: + p = tmp_path / name + p.write_text(text, encoding="utf-8") + return p + + +# --- the four shapes ------------------------------------------------------ + + +def test_a_two_column_csv_is_the_number_shape_and_still_goes_through_parse_csv(tmp_path): + p = _file(tmp_path, "tiles.csv", "Label,Value\nQ3 Revenue,$4.2M\nActive customers,12450\n") + d = intake([p]) + assert d["shape"] == "c" + assert [r["label"] for r in d["rows"]] == ["Q3 Revenue", "Active customers"] + assert d["rows"][0]["expected"] == 4_200_000.0 and d["rows"][0]["raw_value"] == "$4.2M" + assert d["rows"][0]["statement"] is None and d["rows"][0]["question"] is None + + +def test_a_third_column_of_sql_becomes_the_statement_and_the_label_is_not_extended(tmp_path): + p = _file(tmp_path, "tiles.csv", + 'Label,Value,SQL\nQ3 Revenue,$4.2M,"SELECT SUM(total) FROM orders WHERE q = 3"\n') + d = intake([p]) + assert d["shape"] == "d" + (row,) = d["rows"] + # Today `parse_csv` would have read the label as `Q3 Revenue (SELECT SUM(total) ...)`. + assert row["label"] == "Q3 Revenue" + assert row["statement"] == "SELECT SUM(total) FROM orders WHERE q = 3" + assert row["expected"] == 4_200_000.0 + + +def test_a_third_column_that_is_not_sql_is_still_glued_onto_the_label_as_today(tmp_path): + # The one pinned three-column case from `tests/test_reconcile.py`: context, not a statement. + p = _file(tmp_path, "tiles.csv", "Metric,Value,Quarter\nRevenue,4.2M,Q3 2025\n") + d = intake([p]) + (row,) = d["rows"] + assert d["shape"] == "c" and row["label"] == "Revenue (Q3 2025)" and row["statement"] is None + + +def test_plain_lines_are_questions(tmp_path): + p = _file(tmp_path, "questions.txt", + "How many orders did we ship in August?\nWhat was revenue by region last quarter?\n") + d = intake([p]) + assert d["shape"] == "a" + assert [r["question"] for r in d["rows"]] == [ + "How many orders did we ship in August?", "What was revenue by region last quarter?"] + assert all(r["statement"] is None and r["expected"] is None for r in d["rows"]) + + +def test_question_and_sql_pairs_are_the_trusted_sql_shape(tmp_path): + p = _file(tmp_path, "pairs.csv", + 'question,sql\nHow many paid orders?,"SELECT COUNT(*) FROM orders WHERE status = \'paid\'"\n') + d = intake([p]) + assert d["shape"] == "b" + (row,) = d["rows"] + assert row["question"] == "How many paid orders?" + assert row["statement"] == "SELECT COUNT(*) FROM orders WHERE status = 'paid'" + assert row["expected"] is None # produced later, by running the statement + + +def test_sql_pasted_on_its_own_is_a_statement_row_with_no_question_yet(tmp_path): + p = _file(tmp_path, "trusted.sql", + "WITH x AS (SELECT 1 AS n) SELECT SUM(n) FROM x;\n") + d = intake([p]) + assert d["shape"] == "b" + (row,) = d["rows"] + assert row["statement"].startswith("WITH x AS") and row["question"] is None + + +def test_a_json_list_reads_the_same_as_the_csv(tmp_path): + p = _file(tmp_path, "rows.json", json.dumps([ + {"question": "How many paid orders?", "sql": "SELECT COUNT(*) FROM orders WHERE status = 'paid'"}, + {"label": "Q3 Revenue", "value": "$4.2M"}, + "What was revenue by region last quarter?", + ])) + d = intake([p]) + shapes = [r["provenance"]["shape"] for r in d["rows"]] + assert shapes == ["b", "c", "a"] + assert d["rows"][1]["expected"] == 4_200_000.0 + + +def test_the_header_row_is_recognised_by_name_and_never_becomes_a_row(tmp_path): + p = _file(tmp_path, "pairs.csv", "Question,Statement\nHow many?,SELECT COUNT(*) FROM orders\n") + d = intake([p]) + assert len(d["rows"]) == 1 and d["rows"][0]["statement"] == "SELECT COUNT(*) FROM orders" + + +# --- mixed input ------------------------------------------------------------ + + +def test_a_statement_whose_label_matches_a_tile_joins_that_tiles_row(tmp_path): + tiles = _file(tmp_path, "tiles.csv", "Label,Value\nQ3 Revenue,$4.2M\nActive customers,12450\n") + sql = _file(tmp_path, "sql.csv", + 'label,sql\nq3 revenue,"SELECT SUM(total) FROM orders WHERE q = 3"\n' + 'Refund rate,"SELECT AVG(refunded) FROM orders"\n') + d = intake([tiles, sql]) + by_label = {r["label"]: r for r in d["rows"]} + # Matched under the case-and-whitespace fold: the SQL lands on the tile's row. + assert by_label["Q3 Revenue"]["statement"].startswith("SELECT SUM(total)") + assert by_label["Q3 Revenue"]["expected"] == 4_200_000.0 + # An unmatched tile stays a number-only row; an unmatched statement gets its own row. + assert by_label["Active customers"]["statement"] is None + assert by_label["Refund rate"]["expected"] is None and by_label["Refund rate"]["statement"] + assert len(d["rows"]) == 3 + + +def test_provenance_names_the_file_the_line_and_the_persons_words(tmp_path): + p = _file(tmp_path, "tiles.csv", "Label,Value\nQ3 Revenue,$4.2M\n") + d = intake([p], source="the finance dashboard") + (row,) = d["rows"] + assert row["provenance"] == { + "shape": "c", "source": "the finance dashboard", "file": "tiles.csv", "line": 2, + "graded": None, + } + + +# --- what is skipped or refused ---------------------------------------------- + + +def test_a_row_with_nothing_usable_is_skipped_with_a_reason(tmp_path): + p = _file(tmp_path, "tiles.csv", "Label,Value\nStatus,active\nRevenue,100\n") + d = intake([p]) + assert [r["label"] for r in d["rows"]] == ["Revenue"] + (skipped,) = d["skipped"] + assert skipped["line"] == 2 and "number" in skipped["reason"] + + +def test_a_file_with_nothing_usable_exits_four(tmp_path, capsys): + p = _file(tmp_path, "tiles.csv", "Label,Value\nStatus,active\n") + assert reconcile.main(["intake", "--file", str(p)]) == 4 + assert "nothing usable" in capsys.readouterr().err + + +def test_a_missing_file_exits_two(tmp_path, capsys): + assert reconcile.main(["intake", "--file", str(tmp_path / "missing.csv")]) == 2 + assert "not found" in capsys.readouterr().err.lower() + + +def test_the_verb_prints_the_same_json_the_function_returns(tmp_path, capsys): + p = _file(tmp_path, "pairs.csv", 'question,sql\nHow many?,"SELECT COUNT(*) FROM orders"\n') + assert reconcile.main(["intake", "--file", str(p), "--source", "analyst"]) == 0 + printed = json.loads(capsys.readouterr().out) + assert printed == intake([p], source="analyst") + + +def test_a_sql_file_keeps_only_statements_and_names_the_rest(tmp_path): + """A `.sql` file gets the same test a CSV cell gets: a chunk that is not a SELECT or a WITH never + becomes a statement row, so nothing but a read-only statement ever reaches the tier from here.""" + f = tmp_path / "q.sql" + f.write_text("DROP TABLE orders;\nSELECT COUNT(*) FROM orders;\n-- a note\n") + result = intake([f], source=None) + assert [r["statement"] for r in result["rows"]] == ["SELECT COUNT(*) FROM orders"] + assert [s["reason"] for s in result["skipped"]] == ["not a SELECT or WITH statement"] * 2 + + +def test_exit_four_still_prints_what_was_skipped_and_why(tmp_path, capsys): + f = tmp_path / "q.sql" + f.write_text("DROP TABLE orders;") + assert reconcile.main(["intake", "--file", str(f)]) == 4 + out, err = capsys.readouterr() + assert json.loads(out)["skipped"][0]["reason"] == "not a SELECT or WITH statement" + assert "nothing usable" in err diff --git a/tests/test_reconcile_ledger.py b/tests/test_reconcile_ledger.py new file mode 100644 index 00000000..45fe9207 --- /dev/null +++ b/tests/test_reconcile_ledger.py @@ -0,0 +1,821 @@ +"""`reconcile.py ledger` and `reconcile.py findings` — one grade per part of a supplied statement, +and the findings a person reads afterwards. + +The ledger reads fixed filenames in a row directory: what happened when the statement ran, what +`sm prepare` and `sm receipt` said about it, what `sm join-probes` and `sm filter-values judge` +reported, and the probe CSVs the execution tier returned. It applies one set of rules and writes one +grade per part: `confirmed`, `model_gap`, `query_defect` or `unresolved`. The rules have a dependency +in them, and that is the property this file exists to hold still: a join that could not be graded +leaves the fan-out check on its aggregate `unresolved`, said out loud, never silently clean. + +Every fixture below is written by the test itself, in the shapes the real verbs emit. Synthetic +throughout: a `demo` shop over `orders`, `order_items` and `customers`. +""" + +from __future__ import annotations + +import json +import sys +from pathlib import Path + +REPO_ROOT = Path(__file__).resolve().parent.parent +sys.path.insert(0, str(REPO_ROOT / "plugins" / "agami" / "scripts")) + +import reconcile # noqa: E402 +from reconcile import findings, ledger # noqa: E402 + +# --- fixture builders ------------------------------------------------------- + + +def _write(row_dir: Path, name: str, payload) -> None: + row_dir.mkdir(parents=True, exist_ok=True) + text = payload if isinstance(payload, str) else json.dumps(payload) + (row_dir / name).write_text(text, encoding="utf-8") + + +def _ran_ok(row_dir: Path) -> None: + _write(row_dir, "run.json", {"status": "ok", "rule": None, "kind": None, "detail": None}) + + +def _prepare(*aggregates: dict, unchecked: str | None = None) -> dict: + return {"aggregates": list(aggregates), "findings": [f for a in aggregates for f in a["findings"]], + "unchecked": unchecked} + + +def _agg(text: str, status: str, joins: list[str] = (), risks: list[str] = (), + reason: str | None = None) -> dict: + return {"aggregate": text, "scope": "main", "status": status, "joins": list(joins), "reason": reason, + "findings": [{"risk": r, "reason": "x", "triggering_joins": list(joins), "aggregate": text} + for r in risks]} + + +def _receipt(*, tables=(), joins=(), columns=()) -> dict: + return {"tables": {"items": list(tables), "undetermined": None}, + "joins": {"items": list(joins), "undetermined": None}, + "columns": {"items": list(columns), "undetermined": None}} + + +def _table(ref: str, filters: list[dict] = ()) -> dict: + return {"ref": ref, "alias": ref, "qname": f"public.{ref}", "declared": True, "scope": "main", + "filters": list(filters)} + + +def _output(column: str, status: str, source_tables: list[str] | None = None) -> dict: + # The receipt carries the matched metric flattened as `name`, `area`, `expression` and, since + # round 3, the metric's `source_tables`. + return {"kind": "output", "column": column, "scope": "main", "status": status, + "name": "revenue" if status == "matched" else None, + "source_tables": source_tables if status == "matched" else None} + + +def _no_joins() -> dict: + return {"joins": [], "joins_written": 0, "dropped": 0, "cardinality": {}, "unique_by_model": {}, + "unreadable": None} + + +def _no_literals() -> dict: + return {"literals": [], "unreadable": None} + + +def _complete(row_dir: Path) -> None: + """Every file a successful run leaves behind, each saying there was nothing to grade.""" + _ran_ok(row_dir) + _write(row_dir, "statement-prepare.json", _prepare()) + _write(row_dir, "statement-receipt.json", _receipt()) + _write(row_dir, "join-probes.json", _no_joins()) + _write(row_dir, "filter-values.judge.json", _no_literals()) + + +def _join_probe(a: str, ac: str, b: str, bc: str, *, declared_between: bool, matches: bool, + too_big: bool = False) -> dict: + (ta, ca), (tb, cb) = sorted([(a, ac), (b, bc)]) + status = "declared" if matches else ("wrong_key" if declared_between else "undeclared") + keys = [f"{ta}.{ca}", f"{tb}.{cb}"] + probes: dict = {"overlap": [], "cardinality": []} + if status == "undeclared" and not too_big: + probes = {"overlap": [{"from": keys[0], "into": keys[1], "sql": "..."}, + {"from": keys[1], "into": keys[0], "sql": "..."}], + "cardinality": keys} + entry = {"id": "join-1", "endpoints": [a, b], "predicate": f"{a}.{ac} = {b}.{bc}", + "scope": "main", "status": status, "pairs": [[[ta, ca], [tb, cb]]], + "declared_between_tables": declared_between, "written_matches_declared": matches, + "declared_pairs": [[["order_items", "order_id"], ["orders", "id"]]] if declared_between else [], + "too_big_to_probe": too_big, "probes": probes, + "not_probed_because": "a table is over the size guard for probes" if too_big else None, + "declared_cardinality": [], "dropped_rows_probe": None, "dropped_rows_not_emitted_because": None} + if declared_between: + # The sample edge: order_items.order_id -> orders.id, many_to_one, orders the one side. + entry["declared_cardinality"] = [{"relationship": "many_to_one", "from": "order_items", + "to": "orders", "one_side": ["orders"], "matched": matches}] + return entry + + +def _with_dropped_probe(join: dict) -> dict: + left, right = join["endpoints"] + join["dropped_rows_probe"] = {"sql": "...", "left": left, "right": right, "on": join["predicate"]} + return join + + +def _probes(*joins: dict, unique: dict | None = None) -> dict: + return {**_no_joins(), "joins": list(joins), "joins_written": len(joins), + "unique_by_model": unique or {}} + + +def _judged(verdict: str, literal: str = "Paid", column: str = "status") -> dict: + return {"literals": [{"id": "lit-1", "table": "orders", "column": column, "literal": literal, + "tier": "choice_field", "verdict": verdict, "observed": None, + "near_miss": "paid" if verdict == "query_defect" else None, + "op": "=", "rows_with_value": 0, "note": "n"}]} + + +def _parts(result: dict) -> dict[str, dict]: + return {row["part"]: row for row in result["rows"]} + + +# --- the run itself ----------------------------------------------------------- + + +def test_a_statement_that_ran_clean_is_confirmed_on_every_part(tmp_path): + _complete(tmp_path) + _write(tmp_path, "statement-prepare.json", _prepare(_agg("SUM(total)", "not_multiplied"))) + _write(tmp_path, "statement-receipt.json", _receipt( + tables=[_table("orders", [{"expr": "orders.deleted_at IS NULL", "status": "applied"}])], + columns=[_output("total", "matched")])) + result = ledger(tmp_path) + assert _parts(result)["metric:total"]["evidence"] == {"metric": "revenue"} + assert result["verdict"] == "confirmed" + assert {row["verdict"] for row in result["rows"]} == {"confirmed"} + assert set(_parts(result)) == {"runs", "scope", "fan_out:SUM(total)", "aggregation:SUM(total)", + "default_filter:orders:orders.deleted_at IS NULL", "metric:total"} + + +def test_a_statement_the_database_rejected_for_a_missing_column_is_a_query_defect(tmp_path): + _write(tmp_path, "run.json", {"status": "failed", "rule": None, "kind": "column_not_found", + "detail": "d"}) + result = ledger(tmp_path) + assert _parts(result)["runs"]["verdict"] == "query_defect" + assert result["verdict"] == "query_defect" + + +def test_a_scope_refusal_is_a_finding_about_the_semantic_model_not_a_crash(tmp_path): + _write(tmp_path, "run.json", {"status": "refused", "rule": "table_scope", "kind": None, + "detail": "d"}) + parts = _parts(ledger(tmp_path)) + assert parts["scope"]["verdict"] == "model_gap" and parts["scope"]["kind"] == "scope" + # The statement itself is not graded wrong for wanting a table nobody exposed. + assert parts["runs"]["verdict"] == "unresolved" + + +def test_select_star_is_the_persons_defect(tmp_path): + _write(tmp_path, "run.json", {"status": "refused", "rule": "select_star", "kind": None, + "detail": "d"}) + assert _parts(ledger(tmp_path))["runs"]["verdict"] == "query_defect" + + +def test_a_connection_failure_leaves_the_run_unresolved_and_says_so(tmp_path): + _write(tmp_path, "run.json", {"status": "failed", "rule": None, "kind": "auth", "detail": "d"}) + runs = _parts(ledger(tmp_path))["runs"] + assert runs["verdict"] == "unresolved" and "auth" in runs["note"] + + +def test_a_missing_run_record_is_unresolved_never_confirmed(tmp_path): + tmp_path.mkdir(exist_ok=True) + runs = _parts(ledger(tmp_path))["runs"] + assert runs["verdict"] == "unresolved" and "no run record" in runs["note"] + + +# --- joins ---------------------------------------------------------------------- + + +def test_a_join_on_the_declared_key_is_confirmed_without_a_probe(tmp_path): + _ran_ok(tmp_path) + _write(tmp_path, "join-probes.json", {"joins": [ + _join_probe("order_items", "order_id", "orders", "id", declared_between=True, matches=True)], + "unreadable": None}) + parts = _parts(ledger(tmp_path)) + assert parts["join:order_items-orders"]["verdict"] == "confirmed" + # No probe was run, and none was needed; the declaration stands on its own. + assert "join_key:order_items-orders" not in parts + + +def test_a_join_on_a_different_key_than_the_declared_one_is_the_persons_defect(tmp_path): + _ran_ok(tmp_path) + _write(tmp_path, "join-probes.json", {"joins": [ + _join_probe("order_items", "id", "orders", "id", declared_between=True, matches=False)], + "unreadable": None}) + _write(tmp_path, "join-1.overlap.0.csv", "matched\n50\n") # overlap does not rescue it + row = _parts(ledger(tmp_path))["join:order_items-orders"] + assert row["verdict"] == "query_defect" + assert row["evidence"]["declared_pairs"] == [[["order_items", "order_id"], ["orders", "id"]]] + + +def test_an_undeclared_join_whose_keys_overlap_is_a_gap_in_the_semantic_model(tmp_path): + _ran_ok(tmp_path) + _write(tmp_path, "join-probes.json", {"joins": [ + _join_probe("orders", "customer_id", "customers", "id", declared_between=False, matches=False)], + "unreadable": None}) + _write(tmp_path, "join-1.overlap.0.csv", "matched\n50\n") + _write(tmp_path, "join-1.overlap.1.csv", "matched\n0\n") + _write(tmp_path, "cardinality.customers.id.csv", + "total,distinct_count,null_count\n1000,1000,0\n") + _write(tmp_path, "cardinality.orders.customer_id.csv", + "total,distinct_count,null_count\n4000,900,10\n") + parts = _parts(ledger(tmp_path)) + assert parts["join:customers-orders"]["verdict"] == "model_gap" + assert parts["join:customers-orders"]["kind"] == "relationship" + assert parts["join_key:customers-orders"]["verdict"] == "confirmed" + card = parts["cardinality:customers-orders"] + assert card["verdict"] == "confirmed" and card["evidence"]["one_side"] == "customers.id" + + +def test_an_undeclared_join_whose_keys_never_meet_is_the_persons_defect(tmp_path): + _ran_ok(tmp_path) + _write(tmp_path, "join-probes.json", {"joins": [ + _join_probe("orders", "customer_id", "customers", "id", declared_between=False, matches=False)], + "unreadable": None}) + _write(tmp_path, "join-1.overlap.0.csv", "matched\n0\n") + _write(tmp_path, "join-1.overlap.1.csv", "matched\n0\n") + parts = _parts(ledger(tmp_path)) + assert parts["join:customers-orders"]["verdict"] == "query_defect" + assert parts["join_key:customers-orders"]["verdict"] == "query_defect" + + +def test_an_undeclared_join_nobody_probed_is_unresolved(tmp_path): + _ran_ok(tmp_path) + _write(tmp_path, "join-probes.json", {"joins": [ + _join_probe("orders", "customer_id", "customers", "id", declared_between=False, matches=False, + too_big=True)], "unreadable": None}) + parts = _parts(ledger(tmp_path)) + assert parts["join:customers-orders"]["verdict"] == "unresolved" + assert "probe" in parts["join:customers-orders"]["note"] + + +def test_two_repeating_sides_mean_the_join_multiplies_rows(tmp_path): + _ran_ok(tmp_path) + _write(tmp_path, "join-probes.json", {"joins": [ + _join_probe("orders", "customer_id", "customers", "id", declared_between=False, matches=False)], + "unreadable": None}) + _write(tmp_path, "join-1.overlap.0.csv", "matched\n50\n") + _write(tmp_path, "cardinality.customers.id.csv", + "total,distinct_count,null_count\n1000,700,0\n") + _write(tmp_path, "cardinality.orders.customer_id.csv", + "total,distinct_count,null_count\n4000,900,10\n") + assert _parts(ledger(tmp_path))["cardinality:customers-orders"]["verdict"] == "query_defect" + + +# --- aggregates, and the dependency rule ------------------------------------------ + + +def test_a_multiplied_total_over_a_confirmed_join_is_the_persons_defect(tmp_path): + _ran_ok(tmp_path) + _write(tmp_path, "join-probes.json", {"joins": [ + _join_probe("order_items", "order_id", "orders", "id", declared_between=True, matches=True)], + "unreadable": None}) + _write(tmp_path, "statement-prepare.json", _prepare( + _agg("SUM(orders.total)", "multiplied", joins=["order_items → orders"], risks=["fan_trap"]))) + row = _parts(ledger(tmp_path))["fan_out:SUM(orders.total)"] + assert row["verdict"] == "query_defect" and "fan_trap" in row["note"] + + +def test_a_multiplication_the_aggregate_cannot_feel_is_confirmed_with_a_note(tmp_path): + _ran_ok(tmp_path) + _write(tmp_path, "join-probes.json", {"joins": [ + _join_probe("order_items", "order_id", "orders", "id", declared_between=True, matches=True)], + "unreadable": None}) + _write(tmp_path, "statement-prepare.json", _prepare( + _agg("COUNT(DISTINCT orders.id)", "multiplied", joins=["order_items → orders"], + risks=["fan_out_invariant"]))) + row = _parts(ledger(tmp_path))["fan_out:COUNT(DISTINCT orders.id)"] + assert row["verdict"] == "confirmed" and "cannot move" in row["note"] + + +def test_a_fan_out_check_over_an_unresolved_join_is_unresolved_never_clean(tmp_path): + """The dependency rule. `sm prepare` had no cardinality for a join the semantic model does not + declare and nobody probed, so its clean verdict on the total is blind, not clean.""" + _ran_ok(tmp_path) + _write(tmp_path, "join-probes.json", {"joins": [ + _join_probe("orders", "customer_id", "customers", "id", declared_between=False, matches=False, + too_big=True)], "unreadable": None}) + _write(tmp_path, "statement-prepare.json", _prepare( + _agg("SUM(orders.total)", "not_multiplied", joins=["orders → customers"]))) + row = _parts(ledger(tmp_path))["fan_out:SUM(orders.total)"] + assert row["verdict"] == "unresolved" + assert row["depends_on"] == ["join:customers-orders"] + assert "customers" in row["note"] + + +def test_a_preflight_that_could_not_run_leaves_every_aggregate_unresolved(tmp_path): + _ran_ok(tmp_path) + _write(tmp_path, "statement-prepare.json", _prepare(unchecked="sqlglot is not installed")) + _write(tmp_path, "statement-receipt.json", _receipt(columns=[_output("total", "matched")])) + result = ledger(tmp_path) + assert result["verdict"] == "unresolved" + assert any(r["part"] == "fan_out:*" and r["verdict"] == "unresolved" for r in result["rows"]) + + +def test_a_sum_over_a_column_that_cannot_be_summed_is_the_persons_defect(tmp_path): + _ran_ok(tmp_path) + _write(tmp_path, "statement-prepare.json", _prepare( + _agg("SUM(rate)", "not_multiplied", risks=["bad_aggregation"]))) + assert _parts(ledger(tmp_path))["aggregation:SUM(rate)"]["verdict"] == "query_defect" + + +# --- filters, metrics, values ------------------------------------------------------- + + +def test_an_omitted_declared_filter_is_a_gap_of_kind_filter(tmp_path): + _ran_ok(tmp_path) + _write(tmp_path, "statement-receipt.json", _receipt( + tables=[_table("orders", [{"expr": "orders.deleted_at IS NULL", "status": "omitted"}])])) + row = _parts(ledger(tmp_path))["default_filter:orders:orders.deleted_at IS NULL"] + assert row["verdict"] == "model_gap" and row["kind"] == "filter" + + +def test_an_output_that_matches_no_metric_is_a_gap_of_kind_metric(tmp_path): + _ran_ok(tmp_path) + _write(tmp_path, "statement-prepare.json", _prepare(_agg("SUM(total)", "not_multiplied"))) + _write(tmp_path, "statement-receipt.json", _receipt(columns=[_output("total", "unmatched")])) + row = _parts(ledger(tmp_path))["metric:total"] + assert row["verdict"] == "model_gap" and row["kind"] == "metric" + + +def test_a_bare_count_is_exempt_from_the_metric_rule(tmp_path): + _ran_ok(tmp_path) + _write(tmp_path, "statement-prepare.json", _prepare(_agg("COUNT(*)", "not_multiplied"))) + _write(tmp_path, "statement-receipt.json", _receipt(columns=[_output("n", "unmatched")])) + row = _parts(ledger(tmp_path))["metric:n"] + assert row["verdict"] == "confirmed" and "count" in row["note"] + + +def test_a_typed_value_carries_the_judges_grade(tmp_path): + _ran_ok(tmp_path) + _write(tmp_path, "filter-values.judge.json", _judged("query_defect")) + row = _parts(ledger(tmp_path))["literal:orders.status=Paid"] + assert row["verdict"] == "query_defect" and row["evidence"]["near_miss"] == "paid" + + +def test_a_stale_list_of_values_is_a_gap_about_the_column(tmp_path): + _ran_ok(tmp_path) + _write(tmp_path, "filter-values.judge.json", _judged("model_gap", literal="refunded")) + row = _parts(ledger(tmp_path))["literal:orders.status=refunded"] + assert row["verdict"] == "model_gap" and row["kind"] == "description" + + +# --- claims, precedence, idempotence --------------------------------------------- + + +def test_claims_are_read_only_when_asked_and_a_difference_is_named_not_judged(tmp_path): + _ran_ok(tmp_path) + _write(tmp_path, "claims.json", {"claims": [ + {"name": "filter_predicates", "status": "differs", "generated": ["a"], "golden": ["b"]}, + {"name": "date_window", "status": "unknown", "generated": None, "golden": None}, + {"name": "tables", "status": "agrees", "generated": ["orders"], "golden": ["orders"]}], + "gates": [], "gated": False}) + assert "predicates" not in _parts(ledger(tmp_path)) + parts = _parts(ledger(tmp_path, with_claims=True)) + assert parts["predicates"]["verdict"] == "unresolved" and parts["predicates"]["evidence"]["generated"] == ["a"] + assert parts["date_window"]["verdict"] == "unresolved" + + +def test_the_weakest_part_decides_the_verdict(tmp_path): + _ran_ok(tmp_path) + _write(tmp_path, "statement-receipt.json", _receipt( + tables=[_table("orders", [{"expr": "orders.deleted_at IS NULL", "status": "omitted"}])])) + _write(tmp_path, "filter-values.judge.json", _judged("query_defect")) + result = ledger(tmp_path) + # model_gap and query_defect both present: the defect wins, and the counts say what else was there. + assert result["verdict"] == "query_defect" + assert result["counts"]["model_gap"] == 1 and result["counts"]["query_defect"] == 1 + + +def test_the_ledger_is_the_same_when_run_twice(tmp_path): + _ran_ok(tmp_path) + _write(tmp_path, "statement-prepare.json", _prepare(_agg("SUM(total)", "not_multiplied"))) + _write(tmp_path, "statement-receipt.json", _receipt(columns=[_output("total", "matched")])) + assert ledger(tmp_path) == ledger(tmp_path) + + +def test_the_verb_writes_ledger_json_beside_the_inputs(tmp_path, capsys): + _ran_ok(tmp_path) + assert reconcile.main(["ledger", "--row-dir", str(tmp_path)]) == 0 + printed = json.loads(capsys.readouterr().out) + assert json.loads((tmp_path / "ledger.json").read_text()) == printed + + +# --- findings ----------------------------------------------------------------------- + + +def _run_dir(tmp_path: Path, rows: list[dict]) -> Path: + run = tmp_path / "run" + (run / "rows").mkdir(parents=True) + with (run / "rows.jsonl").open("w", encoding="utf-8") as fh: + for row in rows: + fh.write(json.dumps(row) + "\n") + return run + + +def test_the_same_missing_join_in_either_order_is_one_finding(tmp_path): + run = _run_dir(tmp_path, [ + {"row": 1, "question": "q1", "statement": "s1", "expected": 1, "status": "mismatch"}, + {"row": 2, "question": "q2", "statement": "s2", "expected": 2, "status": "mismatch"}]) + for n, (a, ac, b, bc) in ((1, ("orders", "customer_id", "customers", "id")), + (2, ("customers", "id", "orders", "customer_id"))): + d = run / "rows" / str(n) + _ran_ok(d) + _write(d, "join-probes.json", {"joins": [ + _join_probe(a, ac, b, bc, declared_between=False, matches=False)], "unreadable": None}) + _write(d, "join-1.overlap.0.csv", "matched\n50\n") + out = findings(run) + keys = [f["key"] for f in out["findings"]] + assert keys == ["relationship:customers-orders"] + (f,) = out["findings"] + assert f["kind"] == "relationship" and [e["row"] for e in f["evidence"]] == [1, 2] + assert json.loads((run / "findings.json").read_text())["findings"][0]["key"] == keys[0] + + +def test_a_filter_gap_and_a_metric_gap_never_share_a_key(tmp_path): + run = _run_dir(tmp_path, [{"row": 1, "question": "q", "statement": "s", "expected": 1, + "status": "mismatch"}]) + d = run / "rows" / "1" + _ran_ok(d) + _write(d, "statement-prepare.json", _prepare(_agg("SUM(total)", "not_multiplied"))) + _write(d, "statement-receipt.json", _receipt( + tables=[_table("orders", [{"expr": "total > 0", "status": "omitted"}])], + columns=[_output("total", "unmatched")])) + keys = {f["key"] for f in findings(run)["findings"]} + assert keys == {"filter:orders:total > 0", "metric:total"} + + +def test_a_clean_statement_the_ai_got_wrong_is_a_finding_of_kind_example(tmp_path): + run = _run_dir(tmp_path, [{"row": 1, "question": "What is total revenue?", "statement": "s", + "expected": 100, "status": "mismatch", + "claims": {"claims": [{"name": "filter_predicates", "status": "differs", + "generated": [], "golden": ["status <> 'cancelled'"]}]}}]) + d = run / "rows" / "1" + _complete(d) + (f,) = findings(run)["findings"] + assert f["kind"] == "example" and f["key"] == "example:what is total revenue?" + assert f["evidence"][0]["claims"]["claims"][0]["name"] == "filter_predicates" + + +def test_the_persons_defects_are_listed_apart_from_the_findings(tmp_path): + run = _run_dir(tmp_path, [{"row": 1, "question": "q", "statement": "s", "expected": 1, + "status": "expected_doubtful"}]) + d = run / "rows" / "1" + _ran_ok(d) + _write(d, "filter-values.judge.json", _judged("query_defect")) + out = findings(run) + assert out["findings"] == [] + (defect,) = out["query_defects"] + assert defect["row"] == 1 and defect["part"] == "literal:orders.status=Paid" + assert json.loads((run / "query_defects.json").read_text()) == out["query_defects"] + assert set(json.loads((run / "ledger.json").read_text())) == {"1"} + + +def test_the_findings_verb_exits_four_when_there_is_nothing_to_read(tmp_path, capsys): + run = tmp_path / "run" + run.mkdir() + assert reconcile.main(["findings", "--run-dir", str(run)]) == 4 + assert "no rows" in capsys.readouterr().err + + +# --- review fixes: nothing confident from nothing ----------------------------------- + + +def test_an_input_missing_after_a_successful_run_leaves_that_part_open(tmp_path): + """A verb that crashed leaves no file, or a zero-byte one, or one JSON error line. Each is a part + of the statement that was NOT checked, and the ledger says so instead of grading the rest clean.""" + _ran_ok(tmp_path) + result = ledger(tmp_path) + assert result["verdict"] == "unresolved" + opened = {p: row for p, row in _parts(result).items() if p.endswith(":*")} + assert set(opened) == {"fan_out:*", "receipt:*", "join:*", "literal:*"} + assert all(row["evidence"]["problem"] == "was not written" for row in opened.values()) + _write(tmp_path, "join-probes.json", "") + _write(tmp_path, "statement-prepare.json", {"error": "no_model"}) + parts = _parts(ledger(tmp_path)) + assert parts["join:*"]["evidence"]["problem"] == "carries an error (empty_file)" + assert parts["fan_out:*"]["evidence"]["problem"] == "carries an error (no_model)" + + +def test_a_run_that_failed_expects_no_later_files(tmp_path): + _write(tmp_path, "run.json", {"status": "failed", "rule": None, "kind": "timeout", "detail": None}) + assert set(_parts(ledger(tmp_path))) == {"runs"} + + +def test_a_complete_clean_row_has_no_open_part(tmp_path): + _complete(tmp_path) + result = ledger(tmp_path) + assert result["verdict"] == "confirmed" and set(_parts(result)) == {"runs", "scope"} + + +def test_an_output_column_the_receipt_could_not_settle_is_open_not_a_gap(tmp_path): + _ran_ok(tmp_path) + _write(tmp_path, "statement-receipt.json", _receipt(columns=[_output("x", "undetermined")])) + row = _parts(ledger(tmp_path))["metric:x"] + assert row["verdict"] == "unresolved" and row["kind"] is None + + +def test_cardinality_headers_are_read_whatever_their_case(tmp_path): + """One tier upper-cases CSV headers. Read case-sensitively, `TOTAL` fell back to the first column + for every field and a unique key read as a repeating one.""" + _ran_ok(tmp_path) + _write(tmp_path, "join-probes.json", {"joins": [ + _join_probe("orders", "customer_id", "customers", "id", declared_between=False, matches=False)], + "unreadable": None}) + _write(tmp_path, "join-1.overlap.0.csv", "MATCHED\n50\n") + _write(tmp_path, "cardinality.customers.id.csv", "TOTAL,DISTINCT_COUNT,NULL_COUNT\n1000,1000,0\n") + _write(tmp_path, "cardinality.orders.customer_id.csv", "TOTAL,DISTINCT_COUNT,NULL_COUNT\n4000,900,10\n") + parts = _parts(ledger(tmp_path)) + assert parts["join:customers-orders"]["verdict"] == "model_gap" + card = parts["cardinality:customers-orders"] + assert card["verdict"] == "confirmed" and card["evidence"]["one_side"] == "customers.id" + + +def test_a_failed_overlap_probe_beside_a_zero_is_not_a_defect(tmp_path): + _ran_ok(tmp_path) + _write(tmp_path, "join-probes.json", {"joins": [ + _join_probe("orders", "customer_id", "customers", "id", declared_between=False, matches=False)], + "unreadable": None}) + _write(tmp_path, "join-1.overlap.0.csv", "") + _write(tmp_path, "join-1.overlap.1.csv", "matched\n0\n") + parts = _parts(ledger(tmp_path)) + assert parts["join:customers-orders"]["verdict"] == "unresolved" + assert "empty" in parts["join:customers-orders"]["note"] + assert parts["join_key:customers-orders"]["verdict"] == "unresolved" + + +def test_two_joins_between_the_same_tables_are_two_parts(tmp_path): + _ran_ok(tmp_path) + first = _join_probe("orders", "id", "order_items", "id", declared_between=True, matches=False) + second = _join_probe("orders", "id", "order_items", "order_id", declared_between=True, matches=True) + second["id"] = "join-2" + _write(tmp_path, "join-probes.json", {"joins": [first, second], "unreadable": None}) + _write(tmp_path, "statement-prepare.json", _prepare( + _agg("SUM(total)", "not_multiplied", joins=["orders - order_items"]))) + parts = _parts(ledger(tmp_path)) + assert parts["join:order_items-orders"]["verdict"] == "query_defect" + assert parts["join:order_items-orders#2"]["verdict"] == "confirmed" + fan = parts["fan_out:SUM(total)"] + assert fan["verdict"] == "unresolved" and "join:order_items-orders" in fan["depends_on"] + + +def test_a_statement_the_verbs_could_not_read_opens_every_join_and_literal(tmp_path): + _ran_ok(tmp_path) + _write(tmp_path, "join-probes.json", {**_no_joins(), "unreadable": "the statement could not be read"}) + _write(tmp_path, "filter-values.judge.json", {"literals": [], "unreadable": "the statement could not be read"}) + parts = _parts(ledger(tmp_path)) + assert parts["join:*"]["verdict"] == "unresolved" and parts["literal:*"]["verdict"] == "unresolved" + + +def test_a_row_with_an_open_part_or_no_ledger_is_never_an_example(tmp_path): + run = _run_dir(tmp_path, [ + {"row": 1, "question": "q1", "statement": "s1", "expected": 1, "status": "mismatch"}, + {"row": 2, "question": "q2", "statement": "s2", "expected": 2, "status": "mismatch"}, + {"row": 3, "question": "q3", "statement": None, "expected": 3, "status": "mismatch"}]) + _write(run / "rows" / "1", "run.json", {"status": "failed", "rule": None, "kind": "timeout", "detail": None}) + _ran_ok(run / "rows" / "2") # ran, but nothing after it was written: four parts stay open + keys = {f["key"] for f in findings(run)["findings"]} + assert not any(k.startswith("example:") for k in keys), keys + + +def test_one_declared_filter_seen_through_two_aliases_is_one_finding(tmp_path): + run = _run_dir(tmp_path, [ + {"row": 1, "question": "q1", "statement": "s1", "expected": 1, "status": "mismatch"}, + {"row": 2, "question": "q2", "statement": "s2", "expected": 2, "status": "mismatch"}]) + for n, expr in ((1, "o.status != 'cancelled'"), (2, "orders.status != 'cancelled'")): + d = run / "rows" / str(n) + _ran_ok(d) + _write(d, "statement-receipt.json", _receipt( + tables=[_table("orders", [{"expr": expr, "status": "omitted"}])])) + keys = [f["key"] for f in findings(run)["findings"]] + assert keys == ["filter:orders:status != 'cancelled'"] + + +# --- round 3: what the semantic model knows about a declared join reaches the aggregate ------- + + +def test_a_count_through_declared_many_to_one_joins_is_confirmed(tmp_path): + """`COUNT(*)` names no column, so the pre-flight cannot bind it and says `undetermined`. When every + join the statement writes is the declared one and brings in one row at most (its right side is the + one side of the relationship), nothing can multiply the count, and the ledger says so.""" + _ran_ok(tmp_path) + join = _join_probe("order_items", "order_id", "orders", "id", declared_between=True, matches=True) + _write(tmp_path, "join-probes.json", _probes(join, unique={"orders.id": True, "order_items.order_id": False})) + _write(tmp_path, "statement-prepare.json", _prepare( + _agg("COUNT(*)", "undetermined", reason="the aggregate names no column"))) + parts = _parts(ledger(tmp_path)) + assert parts["join:order_items-orders"]["evidence"]["one_row_on_right"] is True + fan = parts["fan_out:COUNT(*)"] + assert fan["verdict"] == "confirmed" and "one row at most" in fan["note"] + assert fan["depends_on"] == ["join:order_items-orders"] + + +def test_the_one_row_stamp_also_comes_from_a_unique_written_column(tmp_path): + """An undeclared join onto a declared key: no relationship to match, but the model says the right + column is unique, which is the same fact.""" + _ran_ok(tmp_path) + join = _join_probe("orders", "customer_id", "customers", "id", declared_between=False, matches=False) + join["probes"] = {"overlap": [], "cardinality": []} + join["status"] = "declared" + _write(tmp_path, "join-probes.json", _probes(join, unique={"customers.id": True, "orders.customer_id": False})) + _write(tmp_path, "statement-prepare.json", _prepare(_agg("COUNT(*)", "undetermined"))) + assert _parts(ledger(tmp_path))["fan_out:COUNT(*)"]["verdict"] == "confirmed" + + +def test_a_join_that_brings_the_many_side_in_leaves_the_count_open_and_names_the_cause(tmp_path): + """`FROM orders JOIN order_items`: the right side is the many side, so each order row can become + several. The count stays open, and the note repeats the pre-flight's reason instead of blaming the join.""" + _ran_ok(tmp_path) + join = _join_probe("orders", "id", "order_items", "order_id", declared_between=True, matches=True) + _write(tmp_path, "join-probes.json", _probes(join, unique={"orders.id": True, "order_items.order_id": False})) + _write(tmp_path, "statement-prepare.json", _prepare( + _agg("COUNT(*)", "undetermined", reason="the aggregate names no column (COUNT(*) counts rows of every joined table)"))) + parts = _parts(ledger(tmp_path)) + assert parts["join:order_items-orders"]["evidence"]["one_row_on_right"] is False + fan = parts["fan_out:COUNT(*)"] + assert fan["verdict"] == "unresolved" + assert "could not bind this aggregate to one table: the aggregate names no column" in fan["note"] + assert fan["evidence"]["reason"].startswith("the aggregate names no column") + + +def test_the_many_to_one_rule_needs_every_join_listed_and_confirmed(tmp_path): + _ran_ok(tmp_path) + good = _join_probe("order_items", "order_id", "orders", "id", declared_between=True, matches=True) + unique = {"orders.id": True, "order_items.order_id": False} + # A join dropped at the cap: one written join is unlisted, so nothing is known about it. + _write(tmp_path, "join-probes.json", {**_probes(good, unique=unique), "joins_written": 2, "dropped": 1}) + _write(tmp_path, "statement-prepare.json", _prepare(_agg("COUNT(*)", "undetermined"))) + assert _parts(ledger(tmp_path))["fan_out:COUNT(*)"]["verdict"] == "unresolved" + # A join on the wrong key beside the good one: the defect blocks the rule. + bad = _join_probe("order_items", "id", "orders", "id", declared_between=True, matches=False) + bad["id"] = "join-2" + _write(tmp_path, "join-probes.json", _probes(good, bad, unique=unique)) + assert _parts(ledger(tmp_path))["fan_out:COUNT(*)"]["verdict"] == "unresolved" + # A self-join has no right side to speak of. + selfjoin = _join_probe("orders", "id", "orders", "id", declared_between=False, matches=False) + selfjoin["status"] = "declared" + selfjoin["probes"] = {"overlap": [], "cardinality": []} + _write(tmp_path, "join-probes.json", _probes(selfjoin, unique={"orders.id": True})) + assert _parts(ledger(tmp_path))["fan_out:COUNT(*)"]["verdict"] == "unresolved" + + +def test_dropped_rows_are_noted_and_never_graded(tmp_path): + """The left table's rows with no partner: a fact the run states. It never decides the verdict, + never blocks an example, and a probe that did not run is noted as such rather than held open.""" + _complete(tmp_path) + join = _with_dropped_probe( + _join_probe("order_items", "order_id", "orders", "id", declared_between=True, matches=True)) + _write(tmp_path, "join-probes.json", _probes(join, unique={"orders.id": True})) + _write(tmp_path, "join-1.dropped_rows.csv", "total,dropped\n4000,12\n") + result = ledger(tmp_path) + noted = _parts(result)["dropped_rows:order_items-orders"] + assert noted["verdict"] == "noted" + assert noted["evidence"] == {"total": 4000, "dropped": 12, "left": "order_items", "right": "orders", "unexamined": None} + assert "12 of 4000 order_items rows have no orders partner" in noted["note"] + assert result["verdict"] == "confirmed" and result["counts"]["noted"] == 1 + _write(tmp_path, "join-1.dropped_rows.csv", "") + noted = _parts(ledger(tmp_path))["dropped_rows:order_items-orders"] + assert noted["verdict"] == "noted" and "nothing is claimed" in noted["note"] + _write(tmp_path, "join-1.dropped_rows.csv", "total,dropped\n4000,0\n") + assert "no order_items row is dropped" in _parts(ledger(tmp_path))["dropped_rows:order_items-orders"]["note"] + + +def test_a_noted_part_does_not_stop_an_example_finding(tmp_path): + run = _run_dir(tmp_path, [{"row": 1, "question": "How many items?", "statement": "s", + "expected": 100, "status": "mismatch"}]) + d = run / "rows" / "1" + _complete(d) + join = _with_dropped_probe( + _join_probe("order_items", "order_id", "orders", "id", declared_between=True, matches=True)) + _write(d, "join-probes.json", _probes(join, unique={"orders.id": True})) + _write(d, "join-1.dropped_rows.csv", "total,dropped\n4000,12\n") + assert [f["kind"] for f in findings(run)["findings"]] == ["example"] + + +# --- round 3: a column nobody declared values for is one gap, said once ----------------------- + + +def _judge_columns(**columns: dict) -> dict: + return {"literals": [], "unreadable": None, + "columns": {key: {"table": key.split(".")[0], "column": key.split(".")[1], "sensitive": False, + "observed_count": None, **fact} for key, fact in columns.items()}} + + +def test_an_undeclared_low_cardinality_column_is_a_gap_in_the_semantic_model(tmp_path): + _ran_ok(tmp_path) + judge = _judge_columns(**{"categories.slug": {"declared": "absent", "distinct": "listed", "observed_count": 8}}) + # Two literals on the column, one part about the column. + judge["literals"] = [{"id": f"lit-{i}", "table": "categories", "column": "slug", "literal": v, "tier": "distinct", + "verdict": "confirmed", "observed": None, "near_miss": None, "op": "=", + "rows_with_value": 3, "note": "n", "declared": "absent"} for i, v in enumerate(("a", "b"), 1)] + _write(tmp_path, "filter-values.judge.json", judge) + parts = _parts(ledger(tmp_path)) + gap = parts["values_declared:categories.slug"] + assert gap["verdict"] == "model_gap" and gap["kind"] == "description" + assert "8 distinct values and the semantic model lists none" in gap["note"] + assert sum(1 for p in parts if p.startswith("values_declared:")) == 1 + assert reconcile._finding_key(gap) == "description:categories.slug" + + +def test_the_values_declared_part_reads_each_state_of_the_column(tmp_path): + _ran_ok(tmp_path) + _write(tmp_path, "filter-values.judge.json", _judge_columns(**{ + "orders.status": {"declared": "populated", "distinct": "not_run"}, + "orders.region": {"declared": "empty", "distinct": "listed", "observed_count": 3}, + "customers.full_name": {"declared": "absent", "distinct": "overflow", "observed_count": 26}, + "orders.notes": {"declared": "absent", "distinct": "not_run"}, + "customers.email": {"declared": "absent", "distinct": "not_run", "sensitive": True}, + })) + parts = _parts(ledger(tmp_path)) + assert parts["values_declared:orders.status"]["verdict"] == "confirmed" + region = parts["values_declared:orders.region"] + assert region["verdict"] == "model_gap" and "nobody decoded" in region["note"] + assert parts["values_declared:customers.full_name"]["verdict"] == "noted" + assert parts["values_declared:orders.notes"]["verdict"] == "unresolved" + assert parts["values_declared:customers.email"]["verdict"] == "noted" + + +# --- round 3: a metric matched by shape on a table the statement never reads ----------------- + + +def test_a_bare_aggregate_matched_to_a_metric_on_another_table_is_open(tmp_path): + _ran_ok(tmp_path) + _write(tmp_path, "statement-receipt.json", _receipt( + tables=[_table("payments")], columns=[_output("total_refunds", "matched", source_tables=["refunds"])])) + row = _parts(ledger(tmp_path))["metric:total_refunds"] + assert row["verdict"] == "unresolved" + assert "defined on refunds, which this statement does not read" in row["note"] + assert row["evidence"]["tables_read"] == ["payments"] + # On the table the metric is defined over, or with no tables named, the match stands. + _write(tmp_path, "statement-receipt.json", _receipt( + tables=[_table("orders")], columns=[_output("revenue", "matched", source_tables=["orders"])])) + assert _parts(ledger(tmp_path))["metric:revenue"]["verdict"] == "confirmed" + _write(tmp_path, "statement-receipt.json", _receipt( + tables=[_table("payments")], columns=[_output("revenue", "matched")])) + assert _parts(ledger(tmp_path))["metric:revenue"]["verdict"] == "confirmed" + + +# --- round 4: the model's spelling, the unexamined side, and the missing states ------------ + + +def test_the_one_row_stamp_reads_the_models_own_spelling_of_a_key(tmp_path): + """The probe file keys `unique_by_model` as the semantic model spells the column; the pair carries + the statement's lowercased spelling. An uppercase-introspected model must still confirm.""" + _ran_ok(tmp_path) + join = _join_probe("orders", "customer_id", "customers", "id", declared_between=False, matches=False) + join["status"] = "declared" + join["probes"] = {"overlap": [], "cardinality": []} + _write(tmp_path, "join-probes.json", _probes(join, unique={"customers.ID": True, "orders.CUSTOMER_ID": False})) + _write(tmp_path, "statement-prepare.json", _prepare(_agg("COUNT(*)", "undetermined"))) + parts = _parts(ledger(tmp_path)) + assert parts["join:customers-orders"]["evidence"]["one_row_on_right"] is True + assert parts["fan_out:COUNT(*)"]["verdict"] == "confirmed" + + +def test_the_unique_column_fallback_decides_a_declared_edge_whose_one_side_is_the_left(tmp_path): + """The live shape of the fallback: a declared relationship matched the written pair, its one + side is the LEFT table, and the right column is nevertheless unique by the semantic model.""" + _ran_ok(tmp_path) + join = _join_probe("orders", "id", "order_items", "order_id", declared_between=True, matches=True) + join["declared_cardinality"] = [{"relationship": "one_to_many", "from": "orders", "to": "order_items", + "one_side": ["orders"], "matched": True}] + _write(tmp_path, "join-probes.json", _probes(join, unique={"order_items.order_id": True, "orders.id": True})) + assert _parts(ledger(tmp_path))["join:order_items-orders"]["evidence"]["one_row_on_right"] is True + + +def test_dropped_rows_read_both_numbers_by_header_and_name_the_side_not_counted(tmp_path): + _complete(tmp_path) + join = _with_dropped_probe( + _join_probe("order_items", "order_id", "orders", "id", declared_between=True, matches=True)) + join["dropped_rows_probe"]["unexamined"] = "orders" + _write(tmp_path, "join-probes.json", _probes(join, unique={"orders.id": True})) + _write(tmp_path, "join-1.dropped_rows.csv", "TOTAL,DROPPED\n4000,12\n") + noted = _parts(ledger(tmp_path))["dropped_rows:order_items-orders"] + assert noted["evidence"]["total"] == 4000 and noted["evidence"]["dropped"] == 12 + assert "rows of orders with no order_items partner were not counted" in noted["note"] + # A one-column result is a probe that did not answer, never N of N dropped. + _write(tmp_path, "join-1.dropped_rows.csv", "dropped\n12\n") + noted = _parts(ledger(tmp_path))["dropped_rows:order_items-orders"] + assert "nothing is claimed" in noted["note"] and "total" not in noted["evidence"] + + +def test_the_values_declared_part_reads_the_remaining_states(tmp_path): + _ran_ok(tmp_path) + _write(tmp_path, "filter-values.judge.json", _judge_columns(**{ + "orders.region": {"declared": "absent", "distinct": "empty", "observed_count": 0}, + "orders.notes": {"declared": "absent", "distinct": "failed"}, + "customers.email": {"declared": "absent", "distinct": "listed", "observed_count": 4, "sensitive": True}, + })) + parts = _parts(ledger(tmp_path)) + assert parts["values_declared:orders.region"]["verdict"] == "noted" + assert parts["values_declared:orders.notes"]["verdict"] == "unresolved" + # A sensitive column is noted before it can be a gap, however few values it holds. + assert parts["values_declared:customers.email"]["verdict"] == "noted" + + +def test_a_metric_over_several_tables_is_confirmed_when_the_statement_reads_one_of_them(tmp_path): + _ran_ok(tmp_path) + _write(tmp_path, "statement-receipt.json", _receipt( + tables=[_table("orders")], columns=[_output("revenue", "matched", source_tables=["orders", "customers"])])) + assert _parts(ledger(tmp_path))["metric:revenue"]["verdict"] == "confirmed"