reconcile: the review panel's fixes

Spec: ACE-128

Three review passes over #304 to #310. Must-fixes: a "same answer, different query" row was never
offered for keep (keep_allowed is the gate's word, whatever the fix); a date window the ledger had
confirmed still read "could not check" and was counted twice (a claim the ledger graded takes the
ledger's word, and a check counts once); the relative-window fold recursed without a depth budget
(eight steps, past which it reads None, so read_claims keeps its promise never to raise). Nits taken:
resume refuses a corrupt checkpoint the way next-chunk does; a definitional claim that could not be
read makes the query fact "not comparable" rather than "same"; a match where agami returned an extra
column stays a match, with the sentence naming the column; a claim that differs between the two
queries is its own state (amber, ≠), so red keeps one meaning; the intake parser refuses a rows file
that is not the intake's output and ties the block to the run named by --out; intake refuses a file
over 20 MB and a CSV cell over the default field limit with one line, never a traceback; the page
shows only the first line of an error; an example is never offered on a row whose ledger holds a
mistake (example_not_offered, read from ledger.json); the fix words live in one place (the items);
the Node page test asserts on CI; three titles lose an em-dash; a tautological assertion, a
duplicated check and two comment periods.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

Author: sandeep-agami

packages/agami-core/src/semantic_model/comparator.py

@@ -653,6 +653,8 @@ def _column_pairs(
             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()))

packages/agami-core/src/semantic_model/golden_claims.py

@@ -470,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.
 
@@ -480,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

plugins/agami/scripts/parse_reconcile_intake.py

@@ -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"])

plugins/agami/scripts/parse_reconcile_report.py

@@ -43,7 +43,27 @@
 _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