AgamiAI · sandeep-agami · Sep 13, 2026 · Sep 13, 2026 · Sep 13, 2026 · Sep 13, 2026
@@ -130,6 +130,10 @@ below corresponds to one such version.
   queries the same) as the pill, and the fix (your query, the semantic model, the examples, the question,
   agami again, nothing) as the action. A query written differently is a noted fact, no longer a blocker
   on a matching answer. (ACE-127)
+- The reconcile card after a second read: the question is the title, columns are compared by the data
+  they carry (the compare-results score names `column_pairs` and `unmatched_generated_columns`), a bare
+  column reads as its table's column in the claims reader, the change text and the decision boxes derive
+  from the one fix, an `example` decision joins the block, the checks panel folds. (ACE-128)
 - From the first test of the grid: a plain column in a list query is no longer graded as a missing
   metric (the receipt's output items say whether they aggregate); a date window written against the
   clock (`date_trunc('year', current_date) + interval`) resolves and compares against another such

@@ -410,6 +410,11 @@ class ItemScore:
     accuracy: Optional[float]
     reason: str
     unmatched_golden_columns: tuple[str, ...] = ()
+    # Which golden column paired with which generated column, by VALUES, and which generated columns
+    # paired with none. A renamed column is the same column here; a reader that compared names would
+    # call it missing. Additive; empty where no column-level comparison ran.
+    column_pairs: tuple[tuple[str, str], ...] = ()
+    unmatched_generated_columns: tuple[str, ...] = ()
     golden_row_count: Optional[int] = None
     generated_row_count: Optional[int] = None
     order_sensitive: Optional[bool] = None
@@ -629,6 +634,32 @@ def _judge(
     return _Verdict("error", None, f"{match!r} is not a match level this comparison knows")
 
 
+def _column_pairs(
+    golden: ExecResult, generated: ExecResult, match: MatchLevel, ordered: bool
+) -> tuple[tuple[tuple[str, str], ...], tuple[str, ...]]:
+    """(golden column, generated column) pairs matched by values, and the generated columns left
+    over, for the two levels that pair columns at all. Never raises; a shape the pairing cannot read
+    reports nothing rather than failing a score that already ran."""
+    if match not in ("exact", "values") or not golden.rows or not generated.rows:
+        return (), ()
+    if len(golden.rows) != len(generated.rows):
+        # Pairing is by value vectors, and two vectors of different length are never equal, so
+        # every column would read unpaired: not a fact about the columns, only about the counts,
+        # which the score already reports. Nothing is claimed here.
+        return (), ()
+    try:
+        pairing, _unmatched = match_columns(
+            golden.columns, golden.rows, generated.columns, generated.rows,
+            ordered=ordered, quantize=match == "values",
+        )
+    except Exception:
+        # The score itself has already reported a ragged or malformed result as an error with a
+        # value-free reason; the pairing is a courtesy on top and must never turn that into a raise.
+        return (), ()
+    pairs = tuple((golden.columns[g], generated.columns[i]) for g, i in sorted(pairing.items()))
+    extra = tuple(name for i, name in enumerate(generated.columns) if i not in set(pairing.values()))
+    return pairs, extra
+
 def compare_result_sets(
     golden: ExecResult,
     generated: ExecResult,
@@ -657,11 +688,14 @@ def compare_result_sets(
         verdict = _Verdict(
             "error", None, f"the comparison failed with an unexpected {type(exc).__name__}"
         )
+    pairs, extra = _column_pairs(golden, generated, match, ordered)
     return ItemScore(
         status=verdict.status,
         accuracy=verdict.accuracy,
         reason=verdict.reason,
         unmatched_golden_columns=verdict.unmatched,
+        column_pairs=pairs,
+        unmatched_generated_columns=extra,
         golden_row_count=len(golden.rows),
         generated_row_count=len(generated.rows),
         order_sensitive=ordered,

@@ -235,6 +235,12 @@ def _rendered(node: "exp.Expression", aliases: dict[str, str], depth: int) -> st
     if isinstance(node, exp.Column):
         qualifier = node.table
         if not qualifier:
+            # An unqualified column in a SELECT that reads exactly one table belongs to that table,
+            # so `opened` and `r.opened` are one key; with two tables in scope it stays bare, since
+            # guessing an owner would make two different columns one key.
+            tables = {rt._tkey(rt._bare(t)) for t in aliases.values()}
+            if len(tables) == 1:
+                return f"{next(iter(tables))}.{node.name.lower()}"
             return node.name.lower()
         # The schema and catalog parts are dropped with the alias: `sales.orders.region` and
         # `orders.region` name one column, and `_bare` has already stripped the schema off the
@@ -464,7 +470,10 @@ def _unit_name(node: "exp.Expression | None") -> Optional[str]:
     return _RELATIVE_UNITS.get(str(text).strip().strip("'\"").lower())
 
 
-def _relative_bound(node: "exp.Expression | None") -> Optional[str]:
+_MAX_RELATIVE_DEPTH = 8
+
+
+def _relative_bound(node: "exp.Expression | None", depth: int = _MAX_RELATIVE_DEPTH) -> Optional[str]:
     """The bound a node spells RELATIVE to the run date, as words: `today`, `start of this year`,
     `start of this year + 7 month`, `today - 30 day`. None for any other shape.
 
@@ -474,36 +483,38 @@ def _relative_bound(node: "exp.Expression | None") -> Optional[str]:
     date it computed would be a bound neither statement wrote. It is compared only against another
     relative bound (`_window_status`), never against a literal date.
     """
-    if node is None:
+    if node is None or depth <= 0:
+        # A relative bound deeper than a handful of steps is not a window anyone wrote; past the
+        # budget it reads None, so `read_claims` keeps its promise never to raise on a pathological tree.
         return None
     if isinstance(node, exp.Paren):
-        return _relative_bound(node.this)
+        return _relative_bound(node.this, depth - 1)
     if isinstance(node, exp.Cast):
-        return _relative_bound(node.this)
+        return _relative_bound(node.this, depth - 1)
     if isinstance(node, exp.CurrentDate):
         return "today"
     if isinstance(node, exp.CurrentTimestamp) or (isinstance(node, exp.Anonymous) and str(node.this).lower() in ("now", "getdate", "sysdate", "current_timestamp")):
         return "now"
     if isinstance(node, (exp.DateTrunc, exp.TimestampTrunc)):
-        inner = _relative_bound(node.this if isinstance(node, exp.TimestampTrunc) else node.args.get("this"))
+        inner = _relative_bound(node.this if isinstance(node, exp.TimestampTrunc) else node.args.get("this"), depth - 1)
         unit = _unit_name(node.args.get("unit"))
         if isinstance(node, exp.DateTrunc):
             # sqlglot's DateTrunc holds the unit in `unit` and the value in `this`; some dialects
             # parse the argument order the other way round, so both are tried.
-            inner = _relative_bound(node.this) or _relative_bound(node.args.get("unit"))
+            inner = _relative_bound(node.this, depth - 1) or _relative_bound(node.args.get("unit"), depth - 1)
             unit = _unit_name(node.args.get("unit")) or _unit_name(node.this)
         if inner in ("today", "now") and unit:
             return f"start of this {unit}"
         return None
     if isinstance(node, (exp.Add, exp.Sub)):
-        base = _relative_bound(node.this)
+        base = _relative_bound(node.this, depth - 1)
         step = _interval_words(node.expression)
         if base and step:
             return _step(base, isinstance(node, exp.Add), *step)
         return None
     date_add_types = tuple(t for t in (exp.DateAdd, getattr(exp, "TsOrDsAdd", None)) if t is not None)
     if isinstance(node, date_add_types + (exp.DateSub,)):
-        base = _relative_bound(node.this)
+        base = _relative_bound(node.this, depth - 1)
         n = node.expression
         unit = _unit_name(node.args.get("unit"))
         count: Optional[int] = None

@@ -30,7 +30,7 @@
 _KEYS = {"profile", "reconcile-run", "intake"}
 
 
-def _key_of(line: str):
+def _key_of(line: str) -> str | None:
     low = line.strip().lower()
     for k in _KEYS:
         if low.startswith(k + ":") or low == k + ":":
@@ -144,8 +144,15 @@ def main(argv=None) -> int:
         print(json.dumps({"ok": False, "data": None, "anomalies": [{"kind": "bad_argument", "detail": str(exc)}],
                           "needs_judgment": {"kind": "bad_argument", "ask": "pass the rows file `reconcile.py intake` wrote and the pasted block"}}, indent=2))
         return 2
+    if not isinstance(intake, dict) or not isinstance(intake.get("rows"), list):
+        print(json.dumps({"ok": False, "data": None, "anomalies": [{"kind": "bad_argument", "detail": "the rows file is not the output of `reconcile.py intake`"}],
+                          "needs_judgment": {"kind": "bad_argument", "ask": "pass the rows file `reconcile.py intake` wrote (an object with a `rows` list)"}}, indent=2))
+        return 2
     known = {r.get("row", n) for n, r in enumerate(intake.get("rows", []), 1)}
-    data, anomalies, needs = parse(text, known, run=args.run)
+    # The block must name the run it is applied to. When --run is not given, the run is the folder
+    # --out lands in, so a block from another run can never be applied by leaving the flag off.
+    run = args.run or (Path(args.out).expanduser().resolve().parent.name if args.out else None)
+    data, anomalies, needs = parse(text, known, run=run)
     counts = None
     if needs is None:
         applied, counts = apply(intake, data["decisions"])

@@ -38,12 +38,32 @@
 from pathlib import Path
 
 _KEYS = {"profile", "reconcile-run", "decisions"}
-_DECISIONS = frozenset({"keep", "change", "fix", "reword", "nothing"})
-_WITH_WORDS = frozenset({"change", "fix", "reword"})
+_DECISIONS = frozenset({"keep", "change", "fix", "reword", "example", "nothing"})
+_WITH_WORDS = frozenset({"change", "fix", "reword", "example"})
 _FIELDS = ("row", "decision", "words")
 _DROPPED_KINDS = frozenset({"unknown_decision", "decision_missing_row", "row_decided_twice",
                             "decision_not_an_object", "keep_not_offered", "words_ignored_on_keep",
-                            "words_ignored_on_nothing", "words_not_text"})
+                            "words_ignored_on_nothing", "words_not_text", "example_not_offered"})
+
+
+def example_blocked_rows(run_dir: Path) -> set[int]:
+    """Rows whose ledger holds a part the data proved wrong: a statement with a mistake in it is never
+    offered as a prompt example, whatever the page suggested. Read from each row's ledger.json."""
+    blocked: set[int] = set()
+    rows_dir = run_dir / "rows"
+    if not rows_dir.is_dir():
+        return blocked
+    for row_dir in rows_dir.iterdir():
+        ledger = row_dir / "ledger.json"
+        if not row_dir.name.isdigit() or not ledger.exists():
+            continue
+        try:
+            parts = json.loads(ledger.read_text(encoding="utf-8")).get("rows", [])
+        except (OSError, ValueError):
+            continue
+        if any(isinstance(p, dict) and p.get("verdict") == "query_defect" for p in parts):
+            blocked.add(int(row_dir.name))
+    return blocked
 
 
 def keepable_rows(run_dir: Path) -> set[int]:
@@ -81,7 +101,7 @@ def keepable_rows(run_dir: Path) -> set[int]:
     return keep
 
 
-def _key_of(line: str):
+def _key_of(line: str) -> str | None:
     low = line.strip().lower()
     for k in _KEYS:
         if low.startswith(k + ":") or low == k + ":":
@@ -112,7 +132,8 @@ def _sections(text: str) -> tuple[dict, list[str]]:
     return out, repeated
 
 
-def parse(text: str, keepable: set[int] | None = None, run: str | None = None) -> tuple[dict, list, dict | None]:
+def parse(text: str, keepable: set[int] | None = None, run: str | None = None,
+          example_blocked: set[int] | None = None) -> tuple[dict, list, dict | None]:
     sec, repeated = _sections(text)
     anomalies: list = [{"kind": "key_repeated", "detail": key} for key in repeated]
     needs: dict | None = None
@@ -156,6 +177,9 @@ def parse(text: str, keepable: set[int] | None = None, run: str | None = None) -
         if row in seen:
             anomalies.append({"kind": "row_decided_twice", "row": row})
             continue
+        if decision == "example" and row in (example_blocked or set()):
+            anomalies.append({"kind": "example_not_offered", "row": row})
+            continue  # dropped like a keep the run did not offer; _DROPPED_KINDS carries the kind
         if decision == "keep" and row not in (keepable or set()):
             # The offer's predicate belongs to the ledger: a keep the run's own files do not
             # allow is not the person's to grant from a page.
@@ -200,7 +224,7 @@ def main(argv=None) -> int:
         print(json.dumps({"ok": False, "data": None, "anomalies": [{"kind": "bad_argument", "detail": str(exc)}],
                           "needs_judgment": {"kind": "bad_argument", "ask": "the block file could not be read"}}, indent=2))
         return 2
-    data, anomalies, needs = parse(text, keepable_rows(run_dir), run=run_dir.name)
+    data, anomalies, needs = parse(text, keepable_rows(run_dir), run=run_dir.name, example_blocked=example_blocked_rows(run_dir))
     print(json.dumps({"ok": needs is None, "data": data, "anomalies": anomalies, "needs_judgment": needs}, indent=2))
     return 0