/fix planner and report

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/agents/planner/examples.py

@@ -524,6 +524,94 @@ _EXAMPLE_F = TaskList(
 )
 
 
+# --------------------------------------------------------------------------- #
+# Example G — top-N ranking.
+# "Top 3 product categories by total revenue."
+# Shows: top-N is ONE retrieve_data query — group by the entity, aggregate the
+# measure with an alias, order by that alias, limit N. NEVER a bare
+# order-by-measure + limit (that ranks raw rows, so the same entity can appear
+# twice — observed in production: "top 3 models" returned one model twice).
+# --------------------------------------------------------------------------- #
+
+_EXAMPLE_G = TaskList(
+    plan_id="example_g",
+    goal_restated="Rank product categories by total revenue and return the top 3.",
+    assumptions=[],
+    open_questions=[],
+    tasks=[
+        Task(
+            id="t1",
+            stage="data_understanding",
+            objective="Confirm the sales source exposes category and revenue.",
+            tool_calls=[ToolCall(tool="check_data", args={"source_id": "src_sales"})],
+            expected_output="source_shape",
+            success_criteria=(
+                "Produced the orders table schema; category and revenue columns "
+                "are present."
+            ),
+            depends_on=[],
+            estimated_cost="low",
+        ),
+        Task(
+            id="t2",
+            stage="data_preparation",
+            objective="Aggregate revenue per category, rank descending, keep the top 3.",
+            tool_calls=[
+                ToolCall(
+                    tool="retrieve_data",
+                    args={
+                        "ir": {
+                            "source_id": "src_sales",
+                            "table_id": "t_orders",
+                            "select": [
+                                {"kind": "column", "column_id": "c_category", "alias": "category"},
+                                {
+                                    "kind": "agg",
+                                    "fn": "sum",
+                                    "column_id": "c_revenue",
+                                    "alias": "total_revenue",
+                                },
+                            ],
+                            "group_by": ["c_category"],
+                            "order_by": [{"column_id": "total_revenue", "dir": "desc"}],
+                            "limit": 3,
+                        }
+                    },
+                )
+            ],
+            expected_output="top3_categories",
+            success_criteria=(
+                "Produced at most 3 rows, one distinct category each, ranked by "
+                "total revenue."
+            ),
+            depends_on=["t1"],
+            estimated_cost="low",
+        ),
+    ],
+)
+
+# --------------------------------------------------------------------------- #
+# Example H — infeasible question (see planner.md "When the catalog cannot
+# answer"). "What is our customer churn rate?" against a sales catalog with no
+# subscription/churn data: no task list is forced onto unrelated columns;
+# instead `infeasible_reason` states the gap + the nearest available data.
+# --------------------------------------------------------------------------- #
+
+_EXAMPLE_H = TaskList(
+    plan_id="example_h",
+    goal_restated="Measure the customer churn rate.",
+    assumptions=[],
+    open_questions=[],
+    tasks=[],
+    infeasible_reason=(
+        "The connected source has no churn or subscription-status data — the "
+        "orders table only carries order-level category, revenue, quantity, and "
+        "dates. Nearest available analyses: repeat-purchase behaviour or revenue "
+        "per customer over time."
+    ),
+)
+
+
 EXAMPLES: list[tuple[str, TaskList]] = [
     ("Which product categories drove last quarter's revenue?", _EXAMPLE_A),
     ("How has monthly revenue trended by region this year, and what's unusual?", _EXAMPLE_B),
@@ -531,6 +619,8 @@ EXAMPLES: list[tuple[str, TaskList]] = [
     ("What is the average and total order value per region?", _EXAMPLE_D),
     ("Total revenue for the East and West regions, counting orders of at least 100.", _EXAMPLE_E),
     ("Give me the summary statistics for order revenue and quantity.", _EXAMPLE_F),
+    ("Which 3 product categories have the best revenue performance?", _EXAMPLE_G),
+    ("What is our customer churn rate?", _EXAMPLE_H),
 ]
 
 

src/agents/planner/schemas.py

@@ -58,3 +58,18 @@ class TaskList(BaseModel):
     assumptions: list[str] = Field(default_factory=list)
     open_questions: list[str] = Field(default_factory=list)
     tasks: list[Task] = Field(default_factory=list)
+    # Infeasible sentinel (planner.md "When the catalog cannot answer"): set with
+    # an EMPTY `tasks` list when no catalog column plausibly holds the requested
+    # measure/entity. Explains what is missing and names the nearest available
+    # data. The coordinator renders it as an honest data-gap answer instead of
+    # running the pipeline — the alternative was the planner force-mapping
+    # unrelated columns (observed: `pa` aliased as "revenue").
+    infeasible_reason: str | None = Field(
+        None,
+        description=(
+            "Set ONLY when the question cannot be answered from the catalog: no "
+            "column plausibly holds the requested measure or entity. State what "
+            "is missing and the nearest data that IS available. Leave tasks "
+            "empty when set."
+        ),
+    )

src/agents/planner/validator.py

@@ -61,6 +61,15 @@ class PlannerValidator:
     ) -> None:
         tasks = task_list.tasks
 
+        # Infeasible sentinel (planner.md "When the catalog cannot answer"): an
+        # empty plan carrying `infeasible_reason` is a VALID outcome — the
+        # coordinator renders it as an honest data-gap answer instead of the
+        # planner force-mapping the question onto unrelated columns. A non-empty
+        # plan keeps normal validation and the reason is ignored (a real plan
+        # wins over a hedge).
+        if task_list.infeasible_reason and not tasks:
+            return
+
         # Check 6 — plan non-empty and within the task cap.
         if not tasks:
             raise PlannerValidationError("plan is empty: at least one task is required")

src/agents/refusals.py

@@ -56,6 +56,39 @@ _BLOCKED = {
 }
 
 
+# Data-gap: the planner judged the bound sources cannot answer the question
+# (planner.md "When the catalog cannot answer"). Deterministic wrapper on
+# purpose — the model that declined to plan is not re-asked to prose it up.
+# Keyed on the pipeline's reply_language ("Indonesian"/"English"), not marker
+# detection: the upstream language decision is authoritative here.
+_DATA_GAP = {
+    "en": (
+        "I can't answer that from the data sources connected to this analysis. "
+        "{reason}You can bind a source that holds this data, or ask me what's "
+        "available (try /help or \"what data do I have?\")."
+    ),
+    "id": (
+        "Saya tidak bisa menjawab itu dari sumber data yang terhubung ke "
+        "analisis ini. {reason}Anda bisa menambahkan sumber yang memuat data "
+        "tersebut, atau tanyakan data apa yang tersedia (coba /help atau "
+        "\"data apa yang saya punya?\")."
+    ),
+}
+
+
+def data_gap_message(reason: str | None, reply_language: str | None = None) -> str:
+    """Answer for an infeasible analysis: the bound sources lack the asked-for data.
+
+    `reason` is the planner's `infeasible_reason` (may be None/empty);
+    `reply_language` is the pipeline's detected language ("Indonesian"/"English").
+    """
+    detail = (reason or "").strip()
+    if detail and not detail.endswith((".", "!", "?")):
+        detail += "."
+    lang = "id" if reply_language == "Indonesian" else "en"
+    return _DATA_GAP[lang].format(reason=f"{detail} " if detail else "")
+
+
 def out_of_scope_message(message: str) -> str:
     """Refusal for a benign but out-of-scope request (the `out_of_scope` intent)."""
     return _OUT_OF_SCOPE["id" if _is_indonesian(message) else "en"]

src/agents/slow_path/coordinator.py

@@ -11,13 +11,16 @@ See AGENT_ARCHITECTURE_CONTEXT_new.md §5.2 / §6.1.
 from __future__ import annotations
 
 from collections.abc import Awaitable, Callable
+from datetime import UTC, datetime
 
 from ...catalog.models import Catalog
 from ..planner.contracts import BusinessContext, ToolRegistry
 from ..planner.inputs import Constraints
+from ..planner.schemas import TaskList
 from ..planner.service import PlannerService
+from ..refusals import data_gap_message
 from .assembler import Assembler
-from .schemas import AssembledOutput
+from .schemas import AnalysisRecord, AssembledOutput
 from .task_runner import TaskRunner
 
 
@@ -54,6 +57,13 @@ class SlowPathCoordinator:
         task_list = await self._planner.plan(
             context, catalog, self._registry, query, constraints, **plan_kw
         )
+        if task_list.infeasible_reason and not task_list.tasks:
+            # Honest data-gap outcome (planner.md "When the catalog cannot
+            # answer"): nothing to execute, and the refusal is deliberately
+            # deterministic — not LLM-prosed. The record carries no tasks, so it
+            # is non-substantive: it can never satisfy the report floor or leak
+            # into a report.
+            return _infeasible_output(task_list, context, reply_language)
         if progress:
             await progress(f"Running {len(task_list.tasks)} analysis steps…")
         run_state = await self._task_runner.run(
@@ -65,3 +75,25 @@ class SlowPathCoordinator:
         return await self._assembler.assemble(
             run_state, context, question=query, reply_language=reply_language, **asm_kw
         )
+
+
+def _infeasible_output(
+    task_list: TaskList, context: BusinessContext, reply_language: str | None
+) -> AssembledOutput:
+    """Build the data-gap answer + a faithful (non-substantive) record."""
+    reason = task_list.infeasible_reason or ""
+    return AssembledOutput(
+        chat_answer=data_gap_message(reason, reply_language),
+        analysis_record=AnalysisRecord(
+            goal_restated=task_list.goal_restated,
+            findings=[],
+            caveats=[reason] if reason else [],
+            data_used=[],
+            open_questions=list(task_list.open_questions),
+            tasks_run=[],
+            results_snapshot={},
+            plan_id=task_list.plan_id,
+            business_context_id=context.project_id,
+            created_at=datetime.now(UTC),
+        ),
+    )

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/config/prompts/planner.md

@@ -21,6 +21,12 @@ only a `TaskList` object that conforms to the provided schema.
    id lookup, so a paraphrased name fails.
 5. **No modeling in v1.** There are no modeling tools. Do not emit `modeling`
    tasks. The product is descriptive/diagnostic only — no predictions, no charts.
+6. **Never re-purpose a column as a different business measure.** A column means
+   what the catalog says it means — do not alias one concept as another to force
+   an answer (e.g. selecting an availability percentage AS "revenue", or a
+   0-1 ratio AS a percentage metric). If no column plausibly holds the measure
+   or entity the question asks about, the plan is **infeasible** — see "When the
+   catalog cannot answer".
 
 # How to plan
 
@@ -68,9 +74,29 @@ only a `TaskList` object that conforms to the provided schema.
 - **success_criteria is a reporting signal**, not a control trigger. State, in
   checkable terms (counts, rates, "produced", "above"/"below"), what a good
   result looks like. It never causes a retry.
-- **Surface uncertainty, don't guess.** If the question is ambiguous or the
-  catalog can't fully answer it, record it in `open_questions` and plan the best
-  defensible analysis anyway. Record interpretation choices in `assumptions`.
+- **Surface uncertainty, don't guess.** If the question is *ambiguous* — the
+  catalog can answer it but a term needs interpreting (which period, which
+  metric variant) — record the interpretation in `assumptions`, anything
+  unresolved in `open_questions`, and plan the best defensible analysis. This
+  never licenses re-purposing columns: when the requested measure itself is
+  absent from the catalog, the question is not ambiguous, it is **infeasible**
+  (next section).
+
+# When the catalog cannot answer
+
+Some questions ask for a measure or entity the connected sources simply do not
+hold (e.g. "sales revenue" against a maintenance database, "churn rate" with no
+subscription data). For those:
+
+- Return `tasks: []` and set **`infeasible_reason`**: one short paragraph naming
+  (a) what the question needs that no column provides, and (b) the nearest
+  analyses the catalog CAN support, so the user knows what to ask instead.
+- Do NOT emit a plan that maps the question onto semantically unrelated columns
+  just because their types fit — a confidently wrong number is worse than an
+  honest gap.
+- The test: could you point at a specific catalog column whose *meaning* (name,
+  sample values, table context) matches the requested measure? If not,
+  it is infeasible.
 
 # Writing a retrieve_data QueryIR
 
@@ -109,6 +135,14 @@ only a `TaskList` object that conforms to the provided schema.
   select the product column + `sum(revenue)` aliased `total_revenue`, with
   `group_by: ["<product_col_id>"]`,
   `order_by: [{"column_id": "total_revenue", "dir": "desc"}]`, `limit: 3`.
+  This applies to EVERY entity-ranking phrasing — "top/best/worst/highest/lowest
+  N <entities>", "<entities> with the best <measure> performance", "which
+  <entities> perform best" — the unit being ranked is the ENTITY, so the measure
+  MUST be aggregated per entity first (`group_by` the entity column). Ranking
+  raw rows can return the same entity twice, which is never a valid entity
+  ranking. Choose the aggregate by measure type: additive measures (revenue,
+  counts, backlog) → `sum`; ratio/percentage/rate metrics (availability,
+  utilization, scores) → `avg`. Record the choice in `assumptions`.
 
 # Output
 

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/merge.py

@@ -0,0 +1,136 @@
+"""analyze_merge — combine TWO upstream tables on shared keys (KM-608).
+
+The only analytics "family" tool with a SECOND data input. In ONE call it joins
+two already-materialized tables (`data` = LEFT, `data_right` = RIGHT) on one or
+more shared key columns and returns the combined rows. This is what unlocks the
+"which X has BOTH the worst A and the biggest B" question shape: A and B come
+from two separate `retrieve_data` pulls (e.g. PA-by-section and backlog-by-section)
+that must be aligned per X before either can be judged against the other. Without
+a two-input combine the run dies with ColumnNotFoundError because no single tool
+ever sees both metrics.
+
+Pattern A, extended: it takes TWO `"${t<id>}"` placeholders. The invoker
+materializes BOTH into DataFrames before calling this function (no self-fetch);
+the `on` key(s) reference the column aliases the upstream queries produced.
+
+STATUS: compute layer only — takes two already-materialized DataFrames. The
+wrapper layer (the ToolOutput envelope, dual-arg materialization, ToolSpec
+registration) lives in src/tools/invoker.py + registry.py. Keeping compute
+separate from data-fetching keeps this easy to unit-test and stable when wrapped.
+"""
+
+from __future__ import annotations
+
+import pandas as pd
+
+from src.tools.analytics.descriptive import ColumnNotFoundError
+
+# Join types the tool understands. Whitelisted so an unknown `how` fails loudly
+# instead of silently doing the wrong thing. "cross" is deliberately excluded —
+# it ignores `on` and is never the right tool for the "align two metrics" shape.
+SUPPORTED_HOWS = ("inner", "left", "right", "outer")
+
+
+class UnsupportedJoinError(ValueError):
+    """Requested join type is not in SUPPORTED_HOWS (maps to error_code UNSUPPORTED_JOIN)."""
+
+
+def _clean(value: object) -> object:
+    """Coerce a scalar to a JSON-clean Python value.
+
+    An outer/left/right join introduces `NaN` for non-matching rows, and numpy /
+    pandas scalars (numpy.int64, pandas.Timestamp) are not JSON-serializable —
+    normalise all three so the returned rows are clean.
+    """
+    if isinstance(value, pd.Timestamp):
+        return value.isoformat()
+    if value is None:
+        return None
+    try:
+        if pd.isna(value):
+            return None
+    except (TypeError, ValueError):
+        pass  # non-scalar / unhashable — leave as-is
+    if hasattr(value, "item"):
+        return value.item()
+    return value
+
+
+# Prompt-style description read by the Planner to decide WHEN to pick this tool.
+DESCRIPTION = """\
+Summary: Combine TWO upstream tables into one by joining on shared key column(s) \
+(a pandas merge). `data` is the LEFT table, `data_right` is the RIGHT table; `on` \
+is the shared column alias(es) present in BOTH. Returns the combined rows, one \
+per matched key (join type controlled by `how`, default inner).
+
+USE WHEN a question needs TWO different metrics per the SAME entity and those \
+metrics come from two separate pulls — the tell-tale shape is "which X has BOTH \
+A and B" (e.g. "which section has the worst PA AND the biggest backlog", "top \
+customers by revenue that also have the most complaints"). Plan it as two \
+retrieve_data tasks (one per metric, each keyed by X), then analyze_merge on X.
+
+SETTING KEYS: `on` must be column alias(es) that exist in BOTH tables (the entity \
+you align on, e.g. section_id). Use `suffixes` (default ["_left","_right"]) to \
+disambiguate non-key columns that share a name across the two tables. `how`: \
+inner (only matched keys), left/right (keep one side), outer (keep all).
+
+DON'T USE WHEN:
+  - both metrics can be pulled in ONE retrieve_data query -> just retrieve_data
+  - it groups/aggregates a single table -> analyze_aggregate
+
+Example questions:
+  - "which section has the worst PA and the biggest maintenance backlog"
+  - "regions in the top 10 for sales that are also bottom 10 for margin"
+  - "products low on stock that also have high demand"
+"""
+
+
+def analyze_merge(
+    df: pd.DataFrame,
+    data_right: pd.DataFrame,
+    on: list[str] | str,
+    how: str = "inner",
+    suffixes: tuple[str, str] | list[str] = ("_left", "_right"),
+) -> list[dict[str, object]]:
+    """Join two already-materialized tables on shared key column(s).
+
+    Args:
+        df: LEFT table (in the real system the invoker materializes this from the
+            `data` placeholder).
+        data_right: RIGHT table (materialized from the `data_right` placeholder).
+        on: shared key column alias(es) present in BOTH tables. A bare string is
+            treated as a single key.
+        how: join type — one of SUPPORTED_HOWS (default "inner").
+        suffixes: 2-element (left, right) suffixes applied to non-key columns that
+            collide by name across the two tables.
+
+    Returns:
+        list[dict]: one row per merged record, values JSON-clean (NaN -> None).
+
+    Raises:
+        ColumnNotFoundError: if `on` is empty or a key is absent from either side.
+        UnsupportedJoinError: if `how` is not supported.
+        ValueError: if `suffixes` is not a 2-element sequence.
+    """
+    keys = [on] if isinstance(on, str) else list(on)
+    if not keys:
+        raise ColumnNotFoundError("merge 'on' must name at least one shared key column")
+    if how not in SUPPORTED_HOWS:
+        raise UnsupportedJoinError(
+            f"unsupported join '{how}'; supported: {list(SUPPORTED_HOWS)}"
+        )
+
+    missing_left = [c for c in keys if c not in df.columns]
+    missing_right = [c for c in keys if c not in data_right.columns]
+    if missing_left or missing_right:
+        raise ColumnNotFoundError(
+            f"join key(s) not found — left missing {missing_left}, "
+            f"right missing {missing_right}"
+        )
+
+    suf = tuple(suffixes)
+    if len(suf) != 2:
+        raise ValueError(f"suffixes must be a 2-element (left, right) sequence, got {suffixes!r}")
+
+    merged = df.merge(data_right, on=keys, how=how, suffixes=suf)
+    return [{k: _clean(v) for k, v in rec.items()} for rec in merged.to_dict("records")]

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/tools/invoker.py

@@ -31,6 +31,7 @@ from src.tools.analytics import (
     comparison,
     decomposition,
     descriptive,
+    merge,
     quality,
     relationship,
     segmentation,
@@ -52,6 +53,7 @@ _DISPATCH: dict[str, tuple[Callable[..., Any], str]] = {
     "analyze_correlation": (relationship.analyze_correlation, "stats"),
     "analyze_segment": (segmentation.analyze_segment, "table"),
     "analyze_trend": (temporal.analyze_trend, "series"),
+    "analyze_merge": (merge.analyze_merge, "table"),
 }
 
 
@@ -73,6 +75,19 @@ class AnalyticsToolInvoker:
             return ToolOutput(tool=tool_name, kind="error", error=err)
 
         kwargs = {k: v for k, v in args.items() if k != "data"}
+
+        # Second data input (Pattern A, extended): analyze_merge takes a
+        # `data_right` placeholder the TaskRunner has resolved to another upstream
+        # ToolOutput. Materialize it the same way as `data` and hand the compute fn
+        # a DataFrame, not the raw envelope.
+        if "data_right" in kwargs:
+            df_right, err = _materialize(kwargs["data_right"])
+            if err is not None:
+                err = f"data_right: {err}"
+                logger.warning("tool returned error", tool=tool_name, error=err)
+                return ToolOutput(tool=tool_name, kind="error", error=err)
+            kwargs["data_right"] = df_right
+
         try:
             result = fn(df, **kwargs)
         except Exception as exc:  # noqa: BLE001 — never-throw seam (§8.4)

src/tools/registry.py

@@ -29,6 +29,7 @@ from src.tools.analytics import (
     comparison,
     decomposition,
     descriptive,
+    merge,
     quality,
     relationship,
     segmentation,
@@ -96,6 +97,22 @@ ACTIVE_ANALYTICS_TOOLS: list[ToolSpec] = [
         output_kind="series",
         description=temporal.DESCRIPTION,
     ),
+    ToolSpec(
+        name="analyze_merge",
+        category="analytics.combine",
+        input_schema={
+            "required": ["data", "data_right", "on"],
+            "properties": {
+                "data": {"type": "string"},
+                "data_right": {"type": "string"},
+                "on": {"type": "array"},
+                "how": {"type": "string"},
+                "suffixes": {"type": "array"},
+            },
+        },
+        output_kind="table",
+        description=merge.DESCRIPTION,
+    ),
 ]
 
 # Deferred this round — specs kept intact for easy re-activation, NOT exposed to

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({