[KM-644] harden query path: grouped-select rule, int-date trends, sample decode, trace guard

IRValidator rejects bare selects under group_by (planner retry now
self-corrects the Postgres GroupingError class); analyze_trend handles
integer year/month columns (was collapsing all rows into 1970-01);
catalog/sample_decode.py base64-decodes numeric sample_values at read
(self-disabling stopgap for Go's byte marshaling, 139 values on the live
catalog); traceability skips null db-sources on failed retrievals;
check_data hides -1 row counts. Docs: REPO_STATUS delta + DEV_PLAN §0.5.

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

DEV_PLAN.md

@@ -46,6 +46,26 @@ minter, stream-only). The **Phase 3 traceability build** — scratchpad + `GET /
 
 ---
 
+## 0.5. pr/13 sprint — agent-quality fixes (2026-07-08 live-test review)
+
+Findings from the scoped live sessions (mining analysis, 2026-07-07/08 traces): the planner
+force-mapped absent measures (`pa` aliased as "revenue"), top-N ranked raw rows (duplicate models),
+`analyze_trend` collapsed integer months into a single 1970-01 bucket, an invalid grouped IR reached
+Postgres, failed retrievals wrote all-null traceability sources, and numeric catalog samples arrive
+base64-mangled from Go. Fix tasks (same status legend as §0):
+
+| # | Task | Owner | Status | Note |
+|---|---|---|---|---|
+| Q1 | IR validator: reject bare selects under `group_by` (planner retry self-corrects) | Rifqi | ✅ | `query/ir/validator.py` |
+| Q2 | Planner **infeasible** path: `TaskList.infeasible_reason` + deterministic EN/ID data-gap reply | Rifqi | ✅ | schemas/validator/coordinator/refusals + planner.md "When the catalog cannot answer"; refusal wording → Rifqi to review |
+| Q3 | `analyze_trend`: integer year/month handling (epoch-parse bug) | Rifqi | ✅ | `temporal.py` + 5 local tests |
+| Q4 | Planner few-shots: top-N (Example G) + infeasible (Example H) + entity-vs-row ranking rule | Rifqi | ✅ | live-tested 2026-07-08: backlog top-3 correct via single-IR group+sum; "best PA performance" correct in-process (avg-per-model, assumption recorded). Stale-server trace was a false alarm |
+| Q5 | Catalog numeric `sample_values` base64-decode stopgap (`catalog/sample_decode.py`) | Rifqi | ✅ | self-disabling; **primary fix = Go marshaling — DDL-free handoff to Harry** |
+| Q6 | Traceability null-source suppression + `check_data` `-1` row-count hiding | Rifqi | ✅ | `scratchpad.py` / `data_access.py` |
+| Q7 | `analyze_merge` two-table combine tool (unblocks "worst A + biggest B" questions) | tool owner | ⬜ | flagged out of this sprint; request brief sent by Rifqi |
+| Q8 | Report v2: business-question answer section, unresolved/excluded sections, evidence tables from `results_snapshot`, caveat dedupe, single language | Rifqi/Sofhia | ⬜ | next up; adds one LLM call (button-triggered); new prompt → eval per §7B |
+| Q9 | Record-curation endpoint (`GET …/records` + `exclude_record_ids`) + readiness GET for the FE delta guard | Rifqi ↔ FE | ⬜ | contract addition → API_CONTRACT_BE_PYTHON.md |
+
 ## 1. The direction change (locked decisions from 2026-06-24)
 
 1. **"Problem statement" is replaced by two user-entered fields: `objective` + `business_questions`.**

REPO_STATUS.md

@@ -2,7 +2,7 @@
 
 **Audience:** teammates onboarding onto the Python repo (`Agentic-Service-Data-Eyond-Catalog`).
 **Scope:** what the code does **right now** (branch `pr/4`, ticket KM-652). Describes current state only — no roadmap or to-dos.
-**Snapshot date:** 2026-06-25. **Data-layer reconcile 2026-07-01:** §8/§12 updated — dedorch cutover done, `data_catalog` model reconciled. **Query-path fix 2026-07-02:** §8/§13 — dedorch catalogs ship no FKs → Python infers them (`fk_inference.py`); shared-Fernet-key gotcha documented. **Cross-repo update 2026-06-29:** §2/§8/§11/§12 re-verified against
+**Snapshot date:** 2026-06-25. **Data-layer reconcile 2026-07-01:** §8/§12 updated — dedorch cutover done, `data_catalog` model reconciled. **Query-path fix 2026-07-02:** §8/§13 — dedorch catalogs ship no FKs → Python infers them (`fk_inference.py`); shared-Fernet-key gotcha documented. **Agent-quality fixes 2026-07-08 (pr/13):** from the scoped live-test review — the planner gains an explicit **infeasible** outcome (`TaskList.infeasible_reason` → deterministic EN/ID data-gap reply via `refusals.data_gap_message`; no more force-mapping absent measures like `pa` AS "revenue"), the IR validator rejects bare selects under `group_by` (self-corrects via the planner retry), `analyze_trend` handles integer year/month columns (was collapsing every row into one 1970-01 bucket), planner few-shots add top-N (Example G) + infeasible (Example H), numeric catalog `sample_values` are base64-decoded at read (`catalog/sample_decode.py` — stopgap for Go's byte-marshaling; primary fix is Go-side), traceability no longer emits null source rows for failed retrievals, and `check_data` hides `-1` row counts. **Cross-repo update 2026-06-29:** §2/§8/§11/§12 re-verified against
 the **Go source** (`Orchestrator-Agent-Service`), not its docs. The Go service has moved well past its
 own (uncommitted, stale) design docs: it now hosts the **dedorch SQL migrations** in-repo and a full
 **`/api/v1/analyses` + `/api/v1/skills`** REST surface. Go does **not** call Python yet — those skills

src/catalog/sample_decode.py

@@ -0,0 +1,119 @@
+"""Base64-sample-value decoding for catalogs affected by a dedorch bug.
+
+Go's introspection JSON-marshals numeric sample bytes as base64 (a `[]byte`
+serialization quirk), so every decimal/int-typed column's `sample_values`
+currently arrive as base64 strings (e.g. ``'OTUuMA=='`` for ``"95.0"``)
+instead of the plain numeric text the planner LLM expects. The planner then
+sees gibberish instead of value ranges for exactly the columns it filters and
+aggregates on. Until Go fixes the marshaling, decode these at catalog read
+time.
+
+Conservative by design (a wrong decode silently corrupts planner context):
+  - only numeric-typed columns are considered
+  - every non-null sample in the column must pass a strict base64 gate
+    (valid base64, decodes to printable ASCII, parses as a float) — a single
+    non-conforming entry leaves the WHOLE column untouched (mixed content is
+    suspicious, never guessed)
+  - columns with `sample_values is None` (e.g. PII-flagged columns, which
+    carry no samples by design) are skipped cleanly
+  - self-disabling: once Go ships real numeric samples (plain ``"95.0"`` or
+    actual numbers), the gate fails — plain digit strings are either not
+    base64-padded correctly or don't decode to printable numeric text — so
+    the pass becomes a no-op with no further changes needed here
+"""
+
+from __future__ import annotations
+
+import base64
+import binascii
+
+from src.middlewares.logging import get_logger
+
+from .models import Catalog
+
+logger = get_logger("sample_decode")
+
+_NUMERIC_TYPES = {
+    "int",
+    "integer",
+    "bigint",
+    "decimal",
+    "numeric",
+    "float",
+    "double",
+    "number",
+}
+
+
+def _decode_one(value: str) -> str | None:
+    """Return the decoded numeric text for `value`, or None if it fails the gate."""
+    if len(value) < 2 or len(value) % 4 != 0:
+        return None
+    try:
+        decoded = base64.b64decode(value, validate=True)
+    except (binascii.Error, ValueError):
+        return None
+    try:
+        text = decoded.decode("ascii")
+    except UnicodeDecodeError:
+        return None
+    if not text.isprintable():
+        return None
+    try:
+        float(text)
+    except ValueError:
+        return None
+    return text
+
+
+def _decode_column_samples(samples: list) -> tuple[list, int] | None:
+    """Return (decoded list, count decoded) if every non-null entry passes the gate.
+
+    Returns None if any entry fails the gate (mixed content is left untouched).
+    """
+    decoded_values = []
+    count = 0
+    for entry in samples:
+        if entry is None:
+            decoded_values.append(None)
+            continue
+        if not isinstance(entry, str):
+            return None
+        decoded = _decode_one(entry)
+        if decoded is None:
+            return None
+        count += 1
+        decoded_values.append(decoded)
+    if not count:
+        return None
+    return decoded_values, count
+
+
+def decode_sample_values(catalog: Catalog) -> int:
+    """Decode base64-encoded numeric sample_values in place. Returns count decoded.
+
+    Never raises: any unexpected shape (wrong types, malformed entries) leaves
+    the offending column's values untouched.
+    """
+    total = 0
+    try:
+        for source in catalog.sources:
+            for table in source.tables:
+                for col in table.columns:
+                    if col.data_type.lower() not in _NUMERIC_TYPES:
+                        continue
+                    samples = col.sample_values
+                    if not samples:
+                        continue
+                    result = _decode_column_samples(samples)
+                    if result is None:
+                        continue
+                    decoded_values, count = result
+                    col.sample_values = decoded_values
+                    total += count
+    except Exception as e:
+        logger.error("sample decode failed", error=repr(e))
+        return total
+    if total:
+        logger.info("decoded base64 sample values", user_id=catalog.user_id, count=total)
+    return total

src/catalog/store.py

@@ -15,6 +15,7 @@ from src.middlewares.logging import get_logger
 
 from .fk_inference import infer_foreign_keys
 from .models import Catalog
+from .sample_decode import decode_sample_values
 
 logger = get_logger("catalog_store")
 
@@ -41,7 +42,12 @@ class CatalogStore:
         # dedorch catalogs ship no foreign_keys (Go introspection drops them),
         # but the IR validator only allows FK-backed joins. Infer the obvious
         # edges so the planner and validator agree. No-op once Go emits real FKs.
-        return infer_foreign_keys(Catalog.model_validate(row))
+        catalog = infer_foreign_keys(Catalog.model_validate(row))
+        # dedorch also JSON-marshals numeric sample bytes as base64 (Go bug) —
+        # decode them so the planner sees value ranges, not gibberish.
+        # No-op once Go emits plain numeric samples.
+        decode_sample_values(catalog)
+        return catalog
 
     async def get_by_analysis(self, analysis_id: str) -> Catalog | None:
         """Read the `scope_type='analysis'` catalog row for an analysis.
@@ -63,7 +69,9 @@ class CatalogStore:
             row = result.scalar_one_or_none()
         if row is None:
             return None
-        return infer_foreign_keys(Catalog.model_validate(row))
+        catalog = infer_foreign_keys(Catalog.model_validate(row))
+        decode_sample_values(catalog)
+        return catalog
 
     async def upsert(self, catalog: Catalog) -> None:
         # Legacy: Go's catalog.Service owns catalog writes now. Kept working (and

src/query/ir/validator.py

@@ -87,6 +87,22 @@ class IRValidator:
         for i, col_id in enumerate(ir.group_by):
             self._require_column(columns_by_id, col_id, f"group_by[{i}]")
 
+        # A grouped query must not select bare columns that aren't in group_by —
+        # the database rejects it only at execution ("must appear in the GROUP BY
+        # clause"), which is past the planner's corrective-retry window. Catching
+        # it here turns a failed turn into a self-correcting re-prompt.
+        if ir.group_by:
+            grouped = set(ir.group_by)
+            for i, item in enumerate(ir.select):
+                if item.kind == "column" and item.column_id not in grouped:
+                    raise IRValidationError(
+                        f"select[{i}].column_id {item.column_id!r} is selected bare "
+                        "while group_by is present — every selected column must "
+                        "either appear in group_by or be wrapped in an aggregate "
+                        f'(e.g. {{"kind": "agg", "fn": "sum", '
+                        f'"column_id": {item.column_id!r}}})'
+                    )
+
         for i, ob in enumerate(ir.order_by):
             if ob.column_id not in columns_by_id and ob.column_id not in select_aliases:
                 raise IRValidationError(

src/tools/analytics/temporal.py

@@ -41,6 +41,11 @@ class UnsupportedAggregationError(ValueError):
     """The requested aggregation is not supported (maps to error_code UNSUPPORTED_AGG)."""
 
 
+class InvalidDateColumnError(ValueError):
+    """date_column holds numeric values that aren't a recognizable date/year/month
+    (maps to error_code INVALID_DATE_COLUMN)."""
+
+
 def _clean(value: object) -> object:
     """Convert numpy scalars to plain Python; NaN -> None for JSON-clean output."""
     if value is None:
@@ -53,6 +58,63 @@ def _clean(value: object) -> object:
     return value
 
 
+def _parse_date_column(df: pd.DataFrame, date_column: str) -> pd.Series:
+    """Parse date_column into datetimes, guarding against numeric epoch misparsing.
+
+    pd.to_datetime() treats bare numeric input as epoch-nanoseconds, so bare
+    month numbers (1-12) or calendar years (e.g. 2025) silently collapse to a
+    single 1970 timestamp instead of raising. Numeric columns are resolved
+    explicitly here rather than falling through to pd.to_datetime().
+    """
+    col = df[date_column]
+    if not pd.api.types.is_numeric_dtype(col):
+        return pd.to_datetime(col)
+
+    non_null = col.dropna()
+    is_whole = non_null.empty or (non_null == non_null.astype(int)).all()
+
+    if is_whole and non_null.between(1, 12).all():
+        year_col = next((c for c in df.columns if c.lower() == "year"), None)
+        year_series = df[year_col] if year_col is not None else None
+        year_non_null = year_series.dropna() if year_series is not None else pd.Series(dtype=float)
+        year_ok = (
+            year_series is not None
+            and pd.api.types.is_numeric_dtype(year_series)
+            and not year_non_null.empty
+            and (year_non_null == year_non_null.astype(int)).all()
+            and year_non_null.between(1900, 2100).all()
+        )
+        if not year_ok:
+            raise InvalidDateColumnError(
+                f"date_column '{date_column}' holds bare month numbers (1-12) and no "
+                "'year' column is present in the data — retrieve a year column "
+                "alongside month, or use a real date column."
+            )
+        valid = col.notna() & year_series.notna()
+        result = pd.Series(pd.NaT, index=col.index, dtype="datetime64[ns]")
+        result.loc[valid] = pd.to_datetime(
+            {
+                "year": year_series.loc[valid].astype(int),
+                "month": col.loc[valid].astype(int),
+                "day": 1,
+            }
+        )
+        return result
+
+    if is_whole and non_null.between(1900, 2100).all():
+        result = pd.Series(pd.NaT, index=col.index, dtype="datetime64[ns]")
+        valid = col.notna()
+        result.loc[valid] = pd.to_datetime(
+            col.loc[valid].astype(int).astype(str), format="%Y"
+        )
+        return result
+
+    raise InvalidDateColumnError(
+        f"date_column '{date_column}' is numeric but is not a recognizable date, "
+        "year, or month column."
+    )
+
+
 def _period_label(ts: pd.Timestamp, freq: str) -> str:
     """Human-readable period label keyed off the friendly frequency name."""
     if freq == "month":
@@ -119,6 +181,8 @@ def analyze_trend(
         ColumnNotFoundError: if date_column or value_column is absent.
         InvalidFrequencyError: if freq is not a known period.
         UnsupportedAggregationError: if agg is not supported.
+        InvalidDateColumnError: if date_column is numeric but not a recognizable
+            date, year, or bare month number (needing a companion 'year' column).
     """
     missing = [c for c in (date_column, value_column) if c not in df.columns]
     if missing:
@@ -134,7 +198,7 @@ def analyze_trend(
 
     # Build a clean datetime-indexed series, then resample into periods.
     s = df[[date_column, value_column]].copy()
-    s[date_column] = pd.to_datetime(s[date_column])
+    s[date_column] = _parse_date_column(df, date_column)
     s = s.dropna(subset=[date_column]).set_index(date_column).sort_index()
     resampled = s[value_column].resample(FREQ_MAP[freq]).agg(agg)
 

src/tools/data_access.py

@@ -154,7 +154,9 @@ class DataAccessToolInvoker:
             [
                 t.table_id,
                 t.name,
-                t.row_count,
+                # dedorch catalogs mark an uncounted table as -1; surface None so
+                # the planner prompt never sees a nonsensical "-1 rows".
+                t.row_count if (t.row_count or 0) >= 0 else None,
                 c.column_id,
                 c.name,
                 c.data_type,

src/traceability/scratchpad.py

@@ -124,13 +124,17 @@ class TraceabilityScratchpad:
                 error=out_dict.get("error"),
             )
         )
-        if name == "retrieve_data":
+        if name == "retrieve_data" and status == "success":
             self._record_db_source(output)
 
     def _record_db_source(self, output: Any) -> None:
         # retrieve_data's args are {"ir": ...}; the reliable source_id/table/query
         # live on the tool OUTPUT meta (see tools/data_access.py::_retrieve_data).
         meta = _meta_of(output)
+        if not meta.get("source_id"):
+            # A failed/aborted retrieval carries no provenance meta — emitting it
+            # anyway produced all-null source rows in the payload.
+            return
         query = meta.get("query")
         table = meta.get("table_name") or meta.get("table_id")
         self._db_sources.append({