added column mismatch detector

ai/signatures.py

@@ -160,6 +160,43 @@ class AnalyzeAndPlan(dspy.Signature):
     There is no product_master, products, or product_catalog table.
     Use product_id as the only product identifier. Never invent table names.
 
+    ══════════════════════════════════════════════════════════════
+    RULE 1E — WHICH TABLE OWNS WHICH COLUMNS (DO NOT MIX)
+    ══════════════════════════════════════════════════════════════
+    sales_order_line_pricing  → financial rollup only:
+        gold_amount_per_unit, diamond_amount_per_unit, making_charges_per_unit,
+        base_price_per_unit, selling_price_per_unit, line_total, final_amount,
+        quantity, sol_id, variant_sku, product_id
+        ✗ Does NOT have: gold_kt, gold_colour, gold_rate_per_gm, metal_weight,
+                          diamond_id, shape, quality, pointer, carats
+
+    sales_order_line_gold     → physical gold attributes:
+        gold_kt, gold_colour, gold_rate_per_gm, metal_weight_per_unit,
+        finding_per_unit, gross_weight_per_unit, gold_amount_per_unit, sol_id
+        → JOIN to pricing on sol_id when you need both gold attributes AND costs.
+
+    sales_order_line_diamond  → physical diamond attributes:
+        diamond_id, shape, quality, size_mm, pointer, pieces_per_unit,
+        carats_per_unit, rate_per_carat, diamond_amount_per_unit, sol_id
+        → JOIN to pricing on sol_id ONLY when the question asks about diamond
+          properties (shape, quality, karat, carat) — NOT for cost aggregation.
+
+    RULE: If the question asks "by karat" / "by gold_kt" / "by colour" etc.,
+    you MUST join sales_order_line_gold. You cannot get gold_kt from pricing.
+
+    Example — total costs by karat type:
+      SELECT g.gold_kt,
+             SUM(lp.gold_amount_per_unit    * lp.quantity) AS total_gold_amount,
+             SUM(lp.diamond_amount_per_unit * lp.quantity) AS total_diamond_amount,
+             SUM(lp.making_charges_per_unit * lp.quantity) AS total_making_charges
+      FROM sales_table_v2_sales_order_line_pricing lp
+      JOIN sales_table_v2_sales_order_line_gold g  ON lp.sol_id = g.sol_id
+      JOIN sales_table_v2_sales_order_line     sol ON lp.sol_id = sol.sol_id
+      JOIN sales_table_v2_sales_order          so  ON sol.so_id = so.so_id
+      WHERE so.status = 'closed'
+      GROUP BY g.gold_kt
+      ORDER BY g.gold_kt
+
     ══════════════════════════════════════════════════════════════
     RULE 1A — FAN-OUT: DEDUPLICATE BEFORE AGGREGATING ON JOIN CHAINS
     ══════════════════════════════════════════════════════════════
@@ -436,6 +473,17 @@ class SQLGeneration(dspy.Signature):
 
     4d. NO product_master table — never reference it; use product_id only.
 
+    4e. TABLE COLUMN OWNERSHIP — never use a column from the wrong table:
+        sales_order_line_pricing → has: gold_amount_per_unit, diamond_amount_per_unit,
+          making_charges_per_unit, line_total, quantity, sol_id, variant_sku, product_id
+          ✗ does NOT have: gold_kt, gold_colour, shape, quality, diamond_id
+        sales_order_line_gold → has: gold_kt, gold_colour, gold_rate_per_gm,
+          metal_weight_per_unit (JOIN on sol_id when grouping/filtering by karat or colour)
+        sales_order_line_diamond → has: shape, quality, diamond_id, carats_per_unit
+          (JOIN on sol_id only for property filters, never for cost aggregation)
+        WRONG:  SELECT lp.gold_kt ... FROM sales_order_line_pricing lp
+        CORRECT: JOIN sales_order_line_gold g ON lp.sol_id = g.sol_id, then use g.gold_kt
+
     5. USE PRE-COMPUTED TOTALS — NEVER RECONSTRUCT THEM:
        - For order-level metrics (revenue, AOV): use sales_table_v2_sales_order.total_amount
        - For PO totals: use purchase_orders_v6_purchase_order.total_amount

ai/sql_pattern_checker.py

@@ -9,6 +9,96 @@ import re
 from typing import Any
 
 
+def _build_alias_map(sql: str) -> dict[str, str]:
+    """Extract alias → full_table_name mapping from FROM / JOIN clauses.
+
+    Handles:  FROM table_name alias
+              FROM table_name AS alias
+              JOIN table_name alias
+              JOIN table_name AS alias
+    Returns lower-cased keys and values.
+    """
+    alias_map: dict[str, str] = {}
+    pattern = re.compile(
+        r'(?:FROM|JOIN)\s+"?(\w+)"?\s+(?:AS\s+)?"?(\w+)"?',
+        re.IGNORECASE,
+    )
+    for table, alias in pattern.findall(sql):
+        alias_map[alias.lower()] = table.lower()
+        # also map table → table in case no alias is used
+        alias_map[table.lower()] = table.lower()
+    return alias_map
+
+
+def check_column_table_mismatches(sql: str) -> list[dict[str, Any]]:
+    """Schema-aware check: detect alias.column references where the column
+    does not exist in the aliased table.
+
+    Uses the live database schema so it works for ANY table/column — nothing
+    is hardcoded.  Returns issue dicts in the same format as check_sql_patterns.
+    """
+    try:
+        from db.schema import get_schema
+        schema = get_schema()
+    except Exception:
+        return []   # schema unavailable, skip check
+
+    # Build {table_name_lower: {col_lower, ...}}
+    table_cols: dict[str, set[str]] = {
+        t.lower(): {c["column_name"].lower() for c in cols}
+        for t, cols in schema.items()
+    }
+
+    alias_map = _build_alias_map(sql)
+    issues: list[dict[str, Any]] = []
+    seen: set[str] = set()
+
+    # Find all alias.column references in the SQL
+    for alias, col in re.findall(r'\b(\w+)\.(\w+)\b', sql):
+        alias_l = alias.lower()
+        col_l   = col.lower()
+        key = f"{alias_l}.{col_l}"
+        if key in seen:
+            continue
+        seen.add(key)
+
+        table_l = alias_map.get(alias_l)
+        if table_l is None:
+            continue  # unknown alias (subquery alias, CTE name, etc.) — skip
+        if table_l not in table_cols:
+            continue  # table not in schema — already caught by schema validator
+
+        if col_l not in table_cols[table_l]:
+            # Find which tables DO have this column
+            tables_with_col = [
+                t for t, cols in table_cols.items() if col_l in cols
+            ]
+            # Build a helpful correction hint
+            if tables_with_col:
+                hint = (
+                    f"Column '{col}' does NOT exist in '{table_l}'. "
+                    f"It is available in: {', '.join(tables_with_col)}. "
+                    f"JOIN the correct table on sol_id / so_id / po_id as appropriate "
+                    f"and reference that table's alias instead."
+                )
+            else:
+                hint = (
+                    f"Column '{col}' does NOT exist in '{table_l}' "
+                    f"or any other table in the schema. "
+                    f"Remove it or use a column that actually exists."
+                )
+            issues.append({
+                "pattern_name": f"wrong_table_for_{col_l}",
+                "description": (
+                    f"CRITICAL BUG — column '{col}' referenced via alias '{alias}' "
+                    f"which maps to table '{table_l}', but that column does not exist there."
+                ),
+                "correction": hint,
+            })
+
+    return issues
+
+
 def check_sql_patterns(sql: str) -> list[dict[str, Any]]:
     """Detect known bad patterns in a generated SQL string.
 
@@ -142,6 +232,11 @@ def check_sql_patterns(sql: str) -> list[dict[str, Any]]:
                 ),
             })
 
+    # ── Pattern 4 ────────────────────────────────────────────────────────────
+    # Schema-aware: detect alias.column where column doesn't exist in that table.
+    # Generic — works for gold_kt on pricing table, or any future similar mistake.
+    issues.extend(check_column_table_mismatches(sql))
+
     return issues