fix: prevent NULL inserts for NOT NULL columns (AIRCRAFT/SEAT_CAPACITY_TOTAL pattern)

Two root-cause fixes in generator.py:
1. FK block: `continue` now only fires when fk_ref found AND get_fk_value returns non-None;
previously always fired, leaving unset columns as None if referenced table was empty
2. Coherent entity block: if AI returns null for any column, fall through to generic
generation instead of assigning None (was only handling numeric columns, not type-unknown)

Bridge safety net: convert_value now returns 0.0 instead of None for numeric columns
that receive unparseable strings — satisfies NOT NULL without crashing

Also includes e2e_quality.py improvements: late_complete status, liveboard links
section, --env-name arg, save_summary_md, company name in table display.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

legitdata_bridge.py

@@ -264,12 +264,12 @@ class KeyPairSnowflakeWriter:
                     import re as _re
                     cleaned = _re.sub(r'[^\d.\-+eE]', '', value.replace(',', ''))
                     try:
-                        value = float(cleaned) if cleaned else None
+                        value = float(cleaned) if cleaned else 0.0
                     except (ValueError, TypeError):
-                        return None  # Can't coerce — use NULL rather than crash
+                        value = 0.0  # Can't coerce — use 0 rather than NULL to satisfy NOT NULL
                 elif not isinstance(value, (int, float)):
-                    # dict, list, or other non-numeric type — cannot be cast; use NULL
-                    return None
+                    # dict, list, or other non-numeric type — use 0 rather than NULL
+                    value = 0.0
                 precision = info.get('precision', 38)
                 scale = info.get('scale', 0)
                 if isinstance(value, (int, float)):

legitdata_project/legitdata/generator.py

@@ -991,41 +991,53 @@ class LegitGenerator:
                             fk_ref = fk
                             break
                     if fk_ref:
-                        row[col_name] = self.fk_manager.get_fk_value(
+                        fk_value = self.fk_manager.get_fk_value(
                             fk_ref.references_table,
                             fk_ref.references_column,
                             distribution="pareto" if table.is_fact_table else "uniform"
                         )
-                    continue
-                
+                        if fk_value is not None:
+                            row[col_name] = fk_value
+                            continue
+                        # FK lookup returned None (referenced table empty) — fall through to generic
+                        if i == 0:
+                            print(f"    [WARN] {col_name}: FK lookup returned None for {fk_ref.references_table}.{fk_ref.references_column}, using generic fallback")
+                    # fk_ref not found or FK returned None — fall through to generic
+
                 # Coherent entity columns - get from pre-generated entities
                 if coherent_entities and i < len(coherent_entities):
                     entity = coherent_entities[i]
                     if col_name in entity:
                         entity_value = entity[col_name]
-                        # Check if column is numeric but AI gave us text
-                        data_type_upper = (column.data_type or '').upper()
-                        is_numeric_col = any(t in data_type_upper for t in (
-                            'INT', 'NUMBER', 'NUMERIC', 'DECIMAL', 'BIGINT', 'SMALLINT',
-                            'FLOAT', 'REAL', 'DOUBLE', 'FIXED', 'MONEY',
-                        ))
-
-                        if is_numeric_col:
-                            # Coerce to float and store as Python numeric, not string.
-                            # Strip currency symbols / units before trying.
-                            import re as _re
-                            cleaned = _re.sub(r'[^\d.\-+eE]', '', str(entity_value).replace(',', ''))
-                            try:
-                                row[col_name] = float(cleaned)
-                                continue
-                            except (ValueError, TypeError):
-                                # AI gave us unparseable text — fall through to generic
-                                if i == 0:
-                                    print(f"    [WARN] {col_name}: AI gave '{entity_value}' for numeric column, using inferred instead")
-                                pass  # Fall through to generic generation below
+                        # If AI returned null for this column, fall through to generic generation
+                        if entity_value is None:
+                            if i == 0:
+                                print(f"    [WARN] {col_name}: AI returned null, using generic fallback")
+                            # Fall through to generic generation below
                         else:
-                            row[col_name] = entity_value
-                            continue
+                            # Check if column is numeric but AI gave us text
+                            data_type_upper = (column.data_type or '').upper()
+                            is_numeric_col = any(t in data_type_upper for t in (
+                                'INT', 'NUMBER', 'NUMERIC', 'DECIMAL', 'BIGINT', 'SMALLINT',
+                                'FLOAT', 'REAL', 'DOUBLE', 'FIXED', 'MONEY',
+                            ))
+
+                            if is_numeric_col:
+                                # Coerce to float and store as Python numeric, not string.
+                                # Strip currency symbols / units before trying.
+                                import re as _re
+                                cleaned = _re.sub(r'[^\d.\-+eE]', '', str(entity_value).replace(',', ''))
+                                try:
+                                    row[col_name] = float(cleaned)
+                                    continue
+                                except (ValueError, TypeError):
+                                    # AI gave us unparseable text — fall through to generic
+                                    if i == 0:
+                                        print(f"    [WARN] {col_name}: AI gave '{entity_value}' for numeric column, using inferred instead")
+                                    # Fall through to generic generation below
+                            else:
+                                row[col_name] = entity_value
+                                continue
                 
                 # GENERIC columns - use classification strategy if set, otherwise infer
                 col_class = table_class.get(column.name, {})

tests/e2e_quality.py

@@ -1085,6 +1085,41 @@ def grade_stages(stages: dict, config: dict) -> dict:
     return {"stage_total": total, "breakdown": breakdown}
 
 
+def reconcile_stages_with_logs(stages: dict, diag: dict) -> dict:
+    """
+    If session_logs confirms stages completed that the UI monitor missed
+    (e.g. slow DDL that triggered the 20-min bail-out), upgrade those stages to 'complete'.
+    Returns a new dict — does not mutate the original.
+    """
+    if not diag.get("found"):
+        return stages
+
+    completed_in_logs = set(diag.get("stages_completed", []))
+
+    # Map session_log stage names → UI progress keys
+    log_to_ui = {
+        "research":    "research",
+        "ddl":         "ddl",
+        "populate":    "data",    # session calls it 'populate', UI shows 'Data'
+        "data":        "data",
+        "thoughtspot": "thoughtspot",
+    }
+
+    stages = dict(stages)  # copy — don't mutate
+    for log_stage, ui_key in log_to_ui.items():
+        if log_stage in completed_in_logs and stages.get(ui_key) != "complete":
+            old = stages.get(ui_key, "unknown")
+            stages[ui_key] = "complete"
+            print(f"    ℹ️  Stage '{ui_key}' upgraded to complete via session_logs (monitor saw: {old})")
+
+    # Synthetic 'complete' key — set if all main stages now complete
+    main = ("research", "ddl", "data", "thoughtspot")
+    if all(stages.get(s) == "complete" for s in main):
+        stages["complete"] = "complete"
+
+    return stages
+
+
 def compute_grade(score: float, config: dict) -> str:
     grade = "F"
     for letter, threshold in sorted(config["grading"]["thresholds"].items(), key=lambda x: -x[1]):
@@ -1300,7 +1335,7 @@ def run_single_test(page: Page, test_case: dict, config: dict) -> dict:
         "function": test_case.get("function", ""), "company_url": test_case.get("company_url", ""),
         "stages": {}, "run_context": {}, "stage_grading": {}, "ai_grading": {},
         "total_score": 0.0, "grade": "F",
-        "error": None, "timed_out": False, "duration_seconds": 0,
+        "error": None, "timed_out": False, "late_complete": False, "duration_seconds": 0,
         "diagnostics": {}, "snowflake_check": {},
         "liveboard_viz_count": None,
     }
@@ -1402,8 +1437,16 @@ def run_single_test(page: Page, test_case: dict, config: dict) -> dict:
     else:
         print(f"  📦 Snowflake: schema not found (logged as: {known_schema or 'not in logs'})")
 
-    # Print diagnostics on timeout or error
-    if result["timed_out"] or result["error"]:
+    # Reconcile stages with session_logs — catches cases where the monitor bailed early
+    # but the pipeline actually completed (e.g. slow DDL that took >20 min)
+    result["stages"] = reconcile_stages_with_logs(result["stages"], diag)
+    main_stages = ("research", "ddl", "data", "thoughtspot")
+    if result["timed_out"] and all(result["stages"].get(s) == "complete" for s in main_stages):
+        result["late_complete"] = True
+        print("  ℹ️  Pipeline completed late — all stages confirmed via session_logs")
+
+    # Print diagnostics on timeout/error — skip for late_complete since pipeline did finish
+    if result["error"] or (result["timed_out"] and not result["late_complete"]):
         print_diagnostics(diag, {})  # sf already printed above
 
     # Stage scoring
@@ -1441,10 +1484,85 @@ def save_results(run: dict) -> Path:
     return path
 
 
+def save_summary_md(run: dict, json_path: Path, env_name: str = "") -> Path:
+    """
+    Save a compact markdown summary alongside the JSON.
+    Also writes to latest_{env_name}_summary.md (or latest_summary.md) for nightly use.
+    env_name: 'test' | 'prod' | '' (default)
+    """
+    W = 32
+    lines = [
+        f"# DemoPrep Quality Run — {run['timestamp'][:16]}",
+        f"**Target:** {run.get('target_url', 'unknown')}  |  "
+        f"**Avg:** {run['avg_score']}/100  Grade: {run['overall_grade']}",
+        "",
+        "| Company | Use Case | Data | LB | Total | Note |",
+        "|---------|----------|------|----|-------|------|",
+    ]
+    for r in run["tests"]:
+        ag      = r.get("ai_grading", {})
+        ds      = ag.get("data_score", "n/a")
+        ls      = ag.get("liveboard_score", "n/a")
+        t_str   = f"{r['total_score']}/{r['grade']}"
+        company = r.get("company", r["name"])
+        parts   = [p for p in [r.get("vertical",""), r.get("line",""), r.get("function","")]
+                   if p and p != "* CUSTOM *"]
+        uc      = (" / ".join(parts) if parts else "Custom")[:W]
+        if r.get("error"):            note = "❌ network err"
+        elif r.get("late_complete"):  note = "⚠️ slow (complete)"
+        elif r.get("timed_out"):      note = "⏰ timeout"
+        elif r["grade"] in ("A","B"): note = "🏆 great"
+        elif r["grade"] == "C":       note = "✅ solid"
+        else:                         note = ""
+        ctx     = r.get("run_context") or {}
+        lb_guid = ctx.get("liveboard_guid","")
+        lb_base = (ctx.get("ts_base_url","") or "").rstrip("/")
+        lb_link = f"[lb]({lb_base}/#/pinboard/{lb_guid})" if lb_guid and lb_base else "—"
+        lines.append(f"| {company} | {uc} | {ds} | {ls} | {t_str} {lb_link} | {note} |")
+
+    # Issues and errors
+    issues = []
+    for r in run["tests"]:
+        if r.get("timed_out") and not r.get("late_complete"):
+            last = r.get("diagnostics",{}).get("last_event","unknown")
+            issues.append(f"- **{r.get('company',r['name'])}**: TIMEOUT — last event: {last}")
+        for err in r.get("ai_grading",{}).get("grading_errors",[]):
+            if "skip" not in err.lower():
+                issues.append(f"- **{r.get('company',r['name'])}**: {err}")
+    if issues:
+        lines += ["", "## Issues", ""] + issues
+
+    # Top data weaknesses
+    weaknesses = []
+    for r in run["tests"]:
+        ag = r.get("ai_grading",{})
+        ww = ag.get("data_weaknesses",[])
+        if ww:
+            weaknesses.append(f"**{r.get('company',r['name'])}** (data={ag.get('data_score','?')}/100):")
+            for w in ww[:2]:
+                weaknesses.append(f"  - {w[:120]}")
+    if weaknesses:
+        lines += ["", "## Data Quality Weaknesses", ""] + weaknesses
+
+    lines += ["", "---", f"*JSON: {json_path.name}*"]
+
+    md_text = "\n".join(lines) + "\n"
+    md_path = json_path.with_suffix(".md")
+    md_path.write_text(md_text)
+
+    latest_name = f"latest_{env_name}_summary.md" if env_name else "latest_summary.md"
+    latest = RESULTS_DIR / latest_name
+    latest.write_text(md_text)
+
+    print(f"📋 Summary: {md_path}")
+    print(f"📋 Latest:  {latest}")
+    return md_path
+
+
 # ---------------------------------------------------------------------------
 # Main
 # ---------------------------------------------------------------------------
-def run_quality_suite(max_tests: int = None):
+def run_quality_suite(max_tests: int = None, env_name: str = ""):
     if not TEST_USER or not TEST_PASSWORD:
         raise RuntimeError("TEST_USER and TEST_PASSWORD must be set in .env")
 
@@ -1508,9 +1626,15 @@ def run_quality_suite(max_tests: int = None):
                 viz_n = result.get("liveboard_viz_count")
                 viz_note = f"  ({viz_n} vizzes)" if viz_n is not None else ""
                 print(f"  Liveboard: {ts_url}/#/pinboard/{l_guid}{viz_note}")
+            if result.get("late_complete"):
+                timeout_tag = "  ⚠️  SLOW (completed late)"
+            elif result.get("timed_out"):
+                timeout_tag = "  ⏰ TIMEOUT"
+            else:
+                timeout_tag = ""
             print(f"  TOTAL: {result['total_score']}/100  Grade: {result['grade']}"
                   f"  ({result['duration_seconds']}s)"
-                  f"{'  ⏰ TIMEOUT' if result['timed_out'] else ''}"
+                  f"{timeout_tag}"
                   f"{'  ❌ ERROR' if result['error'] else ''}")
 
         ctx.close()
@@ -1521,10 +1645,12 @@ def run_quality_suite(max_tests: int = None):
 
     run = {
         "run_id": run_id, "timestamp": datetime.now().isoformat(),
+        "target_url": BASE_URL,
         "avg_score": avg, "overall_grade": grade,
         "test_count": len(results), "tests": results,
     }
-    save_results(run)
+    path = save_results(run)
+    save_summary_md(run, path, env_name=env_name)
 
     # --- Summary table ---
     try:
@@ -1561,18 +1687,33 @@ def run_quality_suite(max_tests: int = None):
             lb_url, lb_base = ctx.get("liveboard_guid",""), ctx.get("ts_base_url","")
             lk = _link(f"{lb_base}/#/pinboard/{lb_url}", " 📋 ") if lb_url and lb_base else " —  "
             parts = [p for p in [r.get("vertical",""), r.get("line",""), r.get("function","")] if p and p != "* CUSTOM *"]
-            uc = (" / ".join(parts) if parts else r.get("type","").replace("_"," ").title())[:W_UC]
+            uc = (" / ".join(parts) if parts else r.get("context","")[:W_UC] or "Custom")[:W_UC]
             errs = ag.get("grading_errors", [])
-            if r.get("timed_out"):           note = "⏰ timeout"
-            elif any("parse" in (e or "").lower() for e in errs): note = "❌ parse fail"
-            elif ds == 0 and ls is not None: note = "⚠️ data fail"
-            elif r["grade"] in ("A","B"):    note = "🏆 great"
-            elif r["grade"] == "C":          note = "✅ solid"
-            else:                            note = ""
-            print(_row(r["name"][:W_CO], uc, d_str, l_str, t_str, lk, note))
+            if r.get("error"):                                        note = "❌ network err"
+            elif r.get("late_complete"):                              note = "⚠️ slow"
+            elif r.get("timed_out"):                                  note = "⏰ timeout"
+            elif any("auth failed" in (e or "").lower() for e in errs): note = "❌ auth fail"
+            elif any("parse" in (e or "").lower() for e in errs):    note = "❌ parse fail"
+            elif ds == 0 and ls is not None:                          note = "⚠️ data fail"
+            elif r["grade"] in ("A","B"):                             note = "🏆 great"
+            elif r["grade"] == "C":                                   note = "✅ solid"
+            else:                                                     note = ""
+            company = r.get("company", r["name"])[:W_CO]
+            print(_row(company, uc, d_str, l_str, t_str, lk, note))
 
         print(_div("└", "┴", "┘"))
-        print(f"{'='*62}\n")
+
+        # Liveboard links — plain text for easy copy/click
+        print("\n  Liveboards:")
+        for i, r in enumerate(results, 1):
+            ctx     = r.get("run_context") or {}
+            lb_url  = ctx.get("liveboard_guid", "")
+            lb_base = ctx.get("ts_base_url", "")
+            company = r.get("company", r["name"])
+            url     = f"{lb_base}/#/pinboard/{lb_url}" if lb_url and lb_base else "— not created"
+            print(f"  {i}. {company:<30}  {url}")
+
+        print(f"\n{'='*62}\n")
 
     except Exception as _table_err:
         print(f"\n⚠️  Summary table failed: {_table_err}")
@@ -1601,6 +1742,8 @@ if __name__ == "__main__":
                         help="Fill form but do not click GO — browser opens visibly for inspection")
     parser.add_argument("--url", type=str, default="",
                         help="Override TEST_TARGET_URL (e.g. --url https://thoughtspot-dp-demoprep.hf.space)")
+    parser.add_argument("--env-name", type=str, default="",
+                        help="Tag for summary filename: 'test' → latest_test_summary.md, 'prod' → latest_prod_summary.md")
     args = parser.parse_args()
     if args.dry_run:
         DRY_RUN = True
@@ -1608,4 +1751,7 @@ if __name__ == "__main__":
         BASE_URL = args.url
     if not BASE_URL:
         raise ValueError("No target URL — set TEST_TARGET_URL in .env or pass --url <url>")
-    run_quality_suite(max_tests=args.count or (1 if args.dry_run else None))
+    run_quality_suite(
+        max_tests=args.count or (1 if args.dry_run else None),
+        env_name=args.env_name,
+    )