/fix surface catalog foreign keys to the planner (FK-backed joins)

The slow-path planner's CatalogSummary omitted foreign_keys, but the IR validator
only allows FK-backed joins. So join-requiring questions ("which product made the
most revenue") made the LLM guess joins that were rejected 3x → slow path failed.
The prompt already promised FKs; the rendered catalog just never included them.

- inputs.py: CatalogSummary carries foreign_keys + renders FK lines with the exact
join ids (target_table_id/left_column_id/right_column_id) to copy verbatim.
- planner.md: join ONLY on a listed FK; copy ids; don't invent a join (single
table or say the data isn't linked).
- render.py (single-query planner): FK lines now include the join ids too (were
name-only) — keeps both planners consistent.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

src/agents/planner/inputs.py

@@ -31,11 +31,24 @@ class ColumnSummary(BaseModel):
     top_values: list[Any] | None = None
 
 
+class ForeignKeySummary(BaseModel):
+    """A declared FK edge — the only joins the IR validator accepts.
+
+    Maps directly onto a `retrieve_data` IR join: `column_id` → `left_column_id`,
+    `target_table_id` → `target_table_id`, `target_column_id` → `right_column_id`.
+    """
+
+    column_id: str
+    target_table_id: str
+    target_column_id: str
+
+
 class TableSummary(BaseModel):
     table_id: str
     name: str
     row_count: int | None = None
     columns: list[ColumnSummary] = Field(default_factory=list)
+    foreign_keys: list[ForeignKeySummary] = Field(default_factory=list)
 
 
 class StructuredSourceSummary(BaseModel):
@@ -89,6 +102,16 @@ class CatalogSummary(BaseModel):
                         )
                         for col in table.columns
                     ],
+                    # The declared FKs — the only joins the validator accepts. FKs
+                    # carry no PII (ids only), so they're always surfaced.
+                    foreign_keys=[
+                        ForeignKeySummary(
+                            column_id=fk.column_id,
+                            target_table_id=fk.target_table_id,
+                            target_column_id=fk.target_column_id,
+                        )
+                        for fk in table.foreign_keys
+                    ],
                 )
                 for table in source.tables
             ]
@@ -111,6 +134,12 @@ class CatalogSummary(BaseModel):
         lines: list[str] = []
         for source in self.structured_sources:
             lines.append(f"Source: {source.name} ({source.source_type}) — id={source.source_id}")
+            # Name lookups (within a source) so FK edges render with readable
+            # table/column names alongside the ids the IR join must copy verbatim.
+            table_name_by_id = {t.table_id: t.name for t in source.tables}
+            col_name_by_id = {
+                c.column_id: c.name for t in source.tables for c in t.columns
+            }
             for table in source.tables:
                 rc = f" ({table.row_count:,} rows)" if table.row_count is not None else ""
                 lines.append(f"  Table: {table.name}{rc} — id={table.table_id}")
@@ -121,6 +150,16 @@ class CatalogSummary(BaseModel):
                         f"    - {col.name} [{col.data_type}]: "
                         f"samples={samples}{top} — id={col.column_id}"
                     )
+                for fk in table.foreign_keys:
+                    tgt_table = table_name_by_id.get(fk.target_table_id, fk.target_table_id)
+                    tgt_col = col_name_by_id.get(fk.target_column_id, fk.target_column_id)
+                    src_col = col_name_by_id.get(fk.column_id, fk.column_id)
+                    lines.append(
+                        f"    FK: {src_col} → {tgt_table}.{tgt_col} "
+                        f"(join: target_table_id={fk.target_table_id}, "
+                        f"left_column_id={fk.column_id}, "
+                        f"right_column_id={fk.target_column_id})"
+                    )
             lines.append("")
 
         if self.unstructured_sources:

src/catalog/render.py

@@ -65,5 +65,11 @@ def render_source(source: Source) -> str:
                 tgt_col_name = col_names_by_id.get(fk.target_table_id, {}).get(
                     fk.target_column_id, fk.target_column_id
                 )
-                lines.append(f"    - {src_col_name} -> {tgt_table_name}.{tgt_col_name}")
+                # Include the join ids inline — the planner must copy these verbatim
+                # into the IR join, and the IRValidator does a literal id lookup.
+                lines.append(
+                    f"    - {src_col_name} -> {tgt_table_name}.{tgt_col_name} "
+                    f"(join: target_table_id={fk.target_table_id}, "
+                    f"left_column_id={fk.column_id}, right_column_id={fk.target_column_id})"
+                )
     return "\n".join(lines)

src/config/prompts/planner.md

@@ -41,15 +41,20 @@ only a `TaskList` object that conforms to the provided schema.
   (referencing the upstream result's column aliases).
 - **Measure by a dimension in another table (joins).** When the number you are
   aggregating and the grouping dimension live in DIFFERENT tables of the same
-  database source, add a `joins` entry to the `retrieve_data` IR along a foreign
-  key declared in the catalog — do NOT pick a table that lacks the measure, and do
-  NOT try to "combine" unrelated tables. Example — "revenue by category": the
-  measure `order_items.line_total` joined to `products` on
-  `order_items.product_id = products.id`, grouped by `products.category`. Prefer an
-  existing measure column over recomputing; use a single table (no join) when the
-  measure and dimension already live together (e.g. "revenue by region" from
-  `orders.region` + `orders.total_amount`). Joins are database-only — not available
-  for tabular/file sources.
+  database source, add a `joins` entry to the `retrieve_data` IR. **Join ONLY on a
+  foreign key listed in the catalog.** Each joinable relationship appears as an
+  `FK:` line under its table, e.g.
+  `FK: product_id → products.id (join: target_table_id=t_products, left_column_id=c_oi_product_id, right_column_id=c_products_id)`
+  — copy those three ids verbatim into the join (`target_table_id`,
+  `left_column_id`, `right_column_id`). Example — "revenue by category": the measure
+  `order_items.line_total` joined to `products` on `order_items.product_id =
+  products.id`, grouped by `products.category`. **If no `FK:` line links the tables
+  you need, do NOT invent a join** — the validator rejects any join that isn't a
+  declared FK. Instead use a single table when the measure and dimension already
+  live together (e.g. "revenue by region" from `orders.region` +
+  `orders.total_amount`); if they genuinely aren't linked, say the data isn't
+  connected rather than guessing. Prefer an existing measure column over
+  recomputing. Joins are database-only — not available for tabular/file sources.
 - **Mixing structured + unstructured.** If qualitative context helps, add a
   `retrieve_knowledge` task against an unstructured source listed in the catalog.
 - **CRISP-DM stages.** Tag each task with the stage it serves: