Honor explicit custom dataset contracts

chat_interface.py

@@ -274,11 +274,125 @@ class ChatDemoInterface:
         self.demo_pack_content = ""    # Generated demo pack markdown
         self.spotter_story_ai = ""     # Pure AI-generated Spotter Viz story
         self.spotter_story_matrix = "" # Matrix/ThoughtSpot-recommended Spotter Viz story
+        self.deployment_completion = None  # Final model/liveboard links shown in the right panel
         # Per-session loggers (NOT module-level singletons — avoids cross-session contamination)
         self._session_logger = None
         self._prompt_logger = None
         self._dataset_first_bundle = None
 
+    def clear_deployment_completion(self) -> None:
+        self.deployment_completion = None
+
+    def record_deployment_completion(self, results: dict, database: str, schema_name: str, use_case: str) -> None:
+        """Store final ThoughtSpot artifact links for the UI completion panel."""
+        if not isinstance(results, dict):
+            self.deployment_completion = None
+            return
+
+        ts_url = (self.settings.get('thoughtspot_url') or '').rstrip('/')
+        model_guid = results.get('model_guid') or ''
+        liveboard_guid = results.get('liveboard_guid') or results.get('liveboard_id') or ''
+        liveboard_url = results.get('liveboard_url') or (
+            f"{ts_url}/#/pinboard/{liveboard_guid}" if ts_url and liveboard_guid else ''
+        )
+        model_url = f"{ts_url}/#/data/tables/{model_guid}" if ts_url and model_guid else ''
+        backup = bool(results.get('backup_liveboard')) or results.get('liveboard_creation_path') == 'spotter_tml_backup'
+        warnings = results.get('warnings') or []
+        errors = results.get('errors') or []
+
+        if backup:
+            status = "Backup liveboard created"
+            note = (
+                "MCP/Spotter answer generation was unavailable, so DemoPrep created a clearly marked "
+                "Spotter/TML backup liveboard. The model and data are still available."
+            )
+        elif results.get('success') and liveboard_url:
+            status = "MCP liveboard created"
+            note = "The liveboard was created through the MCP path and enhanced after creation."
+        elif model_url:
+            status = "Model created"
+            note = "The model was created, but a liveboard link was not returned. Use the model link to continue in ThoughtSpot."
+        else:
+            status = "Deployment finished"
+            note = "Review the pipeline status for details."
+
+        self.deployment_completion = {
+            "status": status,
+            "note": note,
+            "model_url": model_url,
+            "liveboard_url": liveboard_url,
+            "model_guid": model_guid,
+            "liveboard_guid": liveboard_guid,
+            "connection": results.get('connection') or '',
+            "schema": f"{database}.{schema_name}" if database and schema_name else schema_name,
+            "use_case": use_case,
+            "warnings": warnings,
+            "errors": errors,
+            "success": bool(results.get('success')),
+        }
+
+    def render_deployment_completion_html(self):
+        """Render final artifact links for the completion panel."""
+        if not self.deployment_completion:
+            return gr.update(value="", visible=False)
+
+        import html as _html
+        item = self.deployment_completion
+        status = _html.escape(str(item.get("status") or "Deployment complete"))
+        note = _html.escape(str(item.get("note") or ""))
+        model_url = str(item.get("model_url") or "")
+        liveboard_url = str(item.get("liveboard_url") or "")
+        schema = _html.escape(str(item.get("schema") or ""))
+        connection = _html.escape(str(item.get("connection") or ""))
+        warnings = item.get("warnings") or []
+        errors = item.get("errors") or []
+        is_mcp = item.get("status") == "MCP liveboard created"
+        border = "#22c55e" if is_mcp and item.get("success") and not item.get("errors") else "#f59e0b"
+        badge_bg = "#dcfce7" if is_mcp else "#fef3c7"
+        badge_color = "#166534" if is_mcp else "#92400e"
+
+        def _link_button(label, url):
+            if not url:
+                return f"<span style='color:#6b7280;font-size:13px;'>{_html.escape(label)} unavailable</span>"
+            safe_url = _html.escape(url, quote=True)
+            return (
+                f"<a href='{safe_url}' target='_blank' rel='noopener noreferrer' "
+                "style='display:inline-block;padding:9px 12px;margin:4px 6px 4px 0;"
+                "border-radius:6px;background:#2563eb;color:white;text-decoration:none;"
+                "font-weight:600;font-size:13px;'>"
+                f"{_html.escape(label)}</a>"
+            )
+
+        warning_html = ""
+        if warnings:
+            warning_items = "".join(f"<li>{_html.escape(str(w))}</li>" for w in warnings[:3])
+            warning_html = f"<ul style='margin:8px 0 0 18px;color:#92400e;font-size:12px;'>{warning_items}</ul>"
+        error_html = ""
+        if errors:
+            error_items = "".join(f"<li>{_html.escape(str(e))}</li>" for e in errors[:2])
+            error_html = f"<ul style='margin:8px 0 0 18px;color:#991b1b;font-size:12px;'>{error_items}</ul>"
+
+        html_value = f"""
+<div style="border:1px solid {border};border-left:5px solid {border};border-radius:8px;padding:12px;margin:10px 0;background:#fff;">
+  <div style="display:flex;align-items:center;justify-content:space-between;gap:8px;">
+    <div style="font-weight:700;color:#111827;">Deployment Links</div>
+    <div style="padding:3px 8px;border-radius:999px;background:{badge_bg};color:{badge_color};font-size:12px;font-weight:700;">{status}</div>
+  </div>
+  <div style="margin-top:8px;color:#374151;font-size:13px;line-height:1.35;">{note}</div>
+  <div style="margin-top:10px;">
+    {_link_button("Open Liveboard", liveboard_url)}
+    {_link_button("Open Model", model_url)}
+  </div>
+  <div style="margin-top:8px;color:#6b7280;font-size:12px;line-height:1.4;">
+    <div><strong>Schema:</strong> {schema or "n/a"}</div>
+    <div><strong>Connection:</strong> {connection or "n/a"}</div>
+  </div>
+  {warning_html}
+  {error_html}
+</div>
+"""
+        return gr.update(value=html_value, visible=True)
+
     def _dataset_first_enabled(self) -> bool:
         """Read dataset-first mode from app config/.env or saved settings."""
         setting = str(self.settings.get("data_generation_mode", "")).strip().lower()
@@ -4326,6 +4440,7 @@ Steps:
                 raise Exception(ts_result["error"])
             
             results = ts_result["results"]
+            self.record_deployment_completion(results, database, schema_name, use_case)
             _ts_meta = {
                 "schema_name": schema_name,
                 "connection": results.get("connection"),
@@ -6223,6 +6338,8 @@ def create_chat_tab(chat_controller_state, settings, current_stage, current_mode
                 elem_classes=["phase-log-stream"],
             )
 
+            deployment_links_panel = gr.HTML(value="", visible=False)
+
             # Timer polls controller.phase_log every 2 seconds (activated when GO is pressed)
             phase_log_timer = gr.Timer(value=2, active=True)
 
@@ -6257,7 +6374,11 @@ def create_chat_tab(chat_controller_state, settings, current_stage, current_mode
                     new_stage = result[1] if len(result) > 1 else stage
                     progress = get_progress_html(new_stage)
                     chatbot_update = gr.update(value=result[0], visible=True)
-                    yield (controller, chatbot_update) + result[1:] + (progress, hide_welcome)
+                    yield (controller, chatbot_update) + result[1:] + (
+                        progress,
+                        hide_welcome,
+                        controller.render_deployment_completion_html(),
+                    )
             except Exception as e:
                 err_tb = traceback.format_exc()
                 print(f"[ERROR] send_message unhandled exception:\n{err_tb}")
@@ -6268,11 +6389,27 @@ def create_chat_tab(chat_controller_state, settings, current_stage, current_mode
                 )
                 history = history or []
                 history.append((message, err_msg))
-                yield (controller, gr.update(value=history, visible=True), stage, model, company, usecase, "", get_progress_html(stage), hide_welcome)
+                completion_update = controller.render_deployment_completion_html() if controller else gr.update(value="", visible=False)
+                yield (
+                    controller,
+                    gr.update(value=history, visible=True),
+                    stage,
+                    model,
+                    company,
+                    usecase,
+                    "",
+                    get_progress_html(stage),
+                    hide_welcome,
+                    completion_update,
+                )
 
         # Wire up send button and enter key
         _send_inputs = [chat_controller_state, msg, chatbot, current_stage, current_model, current_company, current_usecase, ts_env_dropdown, liveboard_name_input]
-        _send_outputs = [chat_controller_state, chatbot, current_stage, current_model, current_company, current_usecase, msg, progress_html, welcome_md]
+        _send_outputs = [
+            chat_controller_state, chatbot, current_stage, current_model,
+            current_company, current_usecase, msg, progress_html, welcome_md,
+            deployment_links_panel,
+        ]
 
         msg.submit(fn=send_message, inputs=_send_inputs, outputs=_send_outputs)
         send_btn.click(fn=send_message, inputs=_send_inputs, outputs=_send_outputs)
@@ -6364,6 +6501,7 @@ def create_chat_tab(chat_controller_state, settings, current_stage, current_mode
                     get_progress_html(stage),
                     hide_welcome,
                     gr.update(open=False),
+                    gr.update(value="", visible=False),
                 )
                 return
 
@@ -6402,6 +6540,7 @@ def create_chat_tab(chat_controller_state, settings, current_stage, current_mode
 
             # Clear phase log for a fresh run
             controller.phase_log = [f"→ GO received — preparing {raw_company} · {use_case_display}"]
+            controller.clear_deployment_completion()
 
             # Tag the run source so session logging can record which interface was used
             controller._run_source = 'app_custom' if is_custom else 'app_defined'
@@ -6419,7 +6558,12 @@ def create_chat_tab(chat_controller_state, settings, current_stage, current_mode
                     new_stage = result[1] if len(result) > 1 else stage
                     progress = get_progress_html(new_stage)
                     chatbot_update = gr.update(value=result[0], visible=True)
-                    yield (controller, chatbot_update) + result[1:5] + (progress, hide_welcome, gr.update(open=False))
+                    yield (controller, chatbot_update) + result[1:5] + (
+                        progress,
+                        hide_welcome,
+                        gr.update(open=False),
+                        controller.render_deployment_completion_html(),
+                    )
             except Exception as e:
                 err_tb = traceback.format_exc()
                 print(f"[ERROR] defined_go unhandled exception:\n{err_tb}")
@@ -6430,7 +6574,18 @@ def create_chat_tab(chat_controller_state, settings, current_stage, current_mode
                     f"The pipeline has been interrupted. You can try again or start a new session."
                 )
                 history.append((f"GO: {use_case_str}", err_msg))
-                yield (controller, gr.update(value=history, visible=True), stage, model, company, usecase, get_progress_html(stage), hide_welcome, gr.update(open=False))
+                yield (
+                    controller,
+                    gr.update(value=history, visible=True),
+                    stage,
+                    model,
+                    company,
+                    usecase,
+                    get_progress_html(stage),
+                    hide_welcome,
+                    gr.update(open=False),
+                    controller.render_deployment_completion_html(),
+                )
 
         _go_inputs = [
             chat_controller_state, vertical_dd, line_dd, function_dd, url_input, use_url_cb,
@@ -6442,7 +6597,7 @@ def create_chat_tab(chat_controller_state, settings, current_stage, current_mode
         _go_outputs = [
             chat_controller_state, chatbot, current_stage, current_model,
             current_company, current_usecase, progress_html, welcome_md,
-            settings_accordion,
+            settings_accordion, deployment_links_panel,
         ]
         go_btn.click(fn=defined_go, inputs=_go_inputs, outputs=_go_outputs)
 

demoprep_app/dataset/explicit_prompt.py

@@ -0,0 +1,404 @@
+"""Build datasets directly from explicit table specs in a custom prompt."""
+
+from __future__ import annotations
+
+import random
+import re
+from dataclasses import dataclass
+from datetime import date, timedelta
+from typing import Any
+
+from demoprep_app.dataset.contracts import DatasetBundle, DatasetColumn, DatasetTable
+from demoprep_app.scenario.contract import ScenarioContract
+
+
+@dataclass(slots=True)
+class ExplicitTableSpec:
+    name: str
+    requested_rows: int | None
+    columns: list[str]
+
+
+def build_explicit_prompt_dataset(
+    *,
+    company_name: str,
+    company_url: str,
+    use_case: str,
+    row_count_guidance: int | None,
+    seed: int,
+) -> DatasetBundle | None:
+    specs = _parse_explicit_table_specs(use_case)
+    if len(specs) < 2 or not any(spec.name.upper().startswith("FACT_") for spec in specs):
+        return None
+
+    rng = random.Random(seed)
+    scenario = ScenarioContract(
+        company_name=company_name,
+        company_url=company_url,
+        use_case=use_case,
+        scenario_type="explicit_table_contract",
+        fact_grain="explicit prompt table grain",
+        demo_audience="Business leaders",
+        business_problem="Analyze the specific tables and metrics requested in the prompt.",
+        dashboard_questions=[
+            "Which metrics need executive attention?",
+            "Where are aging, leakage, or performance gaps concentrated?",
+            "Which customers, regions, or categories are driving the issue?",
+        ],
+        metadata={
+            "seed": seed,
+            "contract_source": "explicit_prompt_tables",
+            "explicit_tables": [spec.name for spec in specs],
+        },
+    )
+
+    tables: list[DatasetTable] = []
+    table_rows: dict[str, list[dict[str, Any]]] = {}
+    primary_keys: dict[str, str] = {}
+
+    for spec in specs:
+        is_fact = spec.name.upper().startswith("FACT_")
+        columns = _columns_for_spec(spec, is_fact=is_fact)
+        row_count = _row_count_for_spec(spec, is_fact=is_fact, row_count_guidance=row_count_guidance)
+        rows = _rows_for_table(spec, columns, row_count, rng, table_rows, primary_keys)
+        table = DatasetTable(spec.name.upper(), _grain_for_table(spec.name), columns, rows, is_fact=is_fact)
+        tables.append(table)
+        table_rows[table.name] = rows
+        if columns:
+            primary_keys[table.name] = columns[0].name
+
+    return DatasetBundle(scenario=scenario, tables=tables)
+
+
+def _parse_explicit_table_specs(text: str) -> list[ExplicitTableSpec]:
+    normalized = (text or "").replace("\u00a0", " ")
+    lines = [line.strip() for line in normalized.splitlines()]
+    specs: list[ExplicitTableSpec] = []
+    table_line_re = re.compile(r"^([A-Za-z][A-Za-z0-9_]*)\s*(?:\((.*?)\))?\s*$")
+
+    idx = 0
+    while idx < len(lines):
+        line = lines[idx]
+        match = table_line_re.match(line)
+        if not match or "_" not in match.group(1):
+            idx += 1
+            continue
+
+        table_name = match.group(1).upper()
+        if not (table_name.startswith("DIM_") or table_name.startswith("FACT_")):
+            idx += 1
+            continue
+
+        requested_rows = _parse_requested_rows(match.group(2) or "")
+        column_line = ""
+        lookahead = idx + 1
+        while lookahead < len(lines):
+            candidate = lines[lookahead].strip()
+            if not candidate:
+                lookahead += 1
+                continue
+            if table_line_re.match(candidate) and "_" in candidate.split()[0]:
+                break
+            column_line = candidate
+            break
+
+        columns = _split_column_line(column_line)
+        if columns:
+            specs.append(ExplicitTableSpec(table_name, requested_rows, columns))
+        idx = max(lookahead + 1, idx + 1)
+
+    return specs
+
+
+def _parse_requested_rows(text: str) -> int | None:
+    match = re.search(r"~?\s*([\d,]+)\s*rows?", text or "", flags=re.IGNORECASE)
+    if not match:
+        return None
+    return int(match.group(1).replace(",", ""))
+
+
+def _split_column_line(line: str) -> list[str]:
+    if not line:
+        return []
+    parts = [part.strip() for part in line.split(",")]
+    columns = []
+    for part in parts:
+        clean = re.sub(r"\s*\(.*?\)", "", part).strip()
+        clean = clean.strip(".")
+        if clean:
+            columns.append(clean)
+    return columns
+
+
+def _columns_for_spec(spec: ExplicitTableSpec, *, is_fact: bool) -> list[DatasetColumn]:
+    columns = []
+    for index, raw_name in enumerate(spec.columns):
+        name = _safe_column_name(raw_name)
+        role = "fact_key" if is_fact and index == 0 else f"{spec.name.lower()}_key" if index == 0 else _semantic_role(name)
+        columns.append(DatasetColumn(name, role, _data_type_for_column(name), nullable=_nullable_for_column(name)))
+
+    existing = {column.name for column in columns}
+    if spec.name.upper() == "FACT_INVOICES":
+        for name, data_type in [
+            ("DAYS_OUTSTANDING", "NUMBER"),
+            ("AGING_BUCKET", "VARCHAR(30)"),
+            ("PAST_DUE_AMOUNT_USD", "NUMBER(14,2)"),
+        ]:
+            if name not in existing:
+                columns.append(DatasetColumn(name, name.lower(), data_type, nullable=False))
+    return columns
+
+
+def _safe_column_name(name: str) -> str:
+    safe = re.sub(r"[^A-Za-z0-9]+", "_", name.strip()).strip("_").upper()
+    aliases = {
+        "INVOICE_CATEGORY_ID": "CATEGORY_ID",
+    }
+    safe = aliases.get(safe, safe)
+    return safe or "COLUMN"
+
+
+def _semantic_role(name: str) -> str:
+    if name.endswith("_ID") or name.endswith("_KEY"):
+        return "dimension_key"
+    if "DATE" in name:
+        return "date"
+    if any(token in name for token in ("AMOUNT", "USD", "DAYS", "REVENUE", "RATE")):
+        return "measure"
+    return "attribute"
+
+
+def _data_type_for_column(name: str) -> str:
+    if "DATE" in name:
+        return "DATE"
+    if any(token in name for token in ("AMOUNT", "USD", "REVENUE", "PAID")):
+        return "NUMBER(14,2)"
+    if name.startswith("DAYS_") or name.endswith("_DAYS"):
+        return "NUMBER"
+    if name.endswith("_ID") or name.endswith("_KEY"):
+        return "NUMBER"
+    return "VARCHAR(120)"
+
+
+def _nullable_for_column(name: str) -> bool:
+    return name in {"PAYMENT_RECEIVED_DATE", "DISPUTE_REASON"}
+
+
+def _row_count_for_spec(spec: ExplicitTableSpec, *, is_fact: bool, row_count_guidance: int | None) -> int:
+    requested = spec.requested_rows or (1000 if is_fact else 50)
+    if is_fact and row_count_guidance:
+        return max(250, min(requested, row_count_guidance))
+    return max(1, requested)
+
+
+def _grain_for_table(table_name: str) -> str:
+    low = table_name.lower()
+    if low.startswith("fact_invoice"):
+        return "invoice"
+    if low.startswith("fact_"):
+        return low.replace("fact_", "").replace("_", " ")
+    return "dimension"
+
+
+def _rows_for_table(
+    spec: ExplicitTableSpec,
+    columns: list[DatasetColumn],
+    row_count: int,
+    rng: random.Random,
+    table_rows: dict[str, list[dict[str, Any]]],
+    primary_keys: dict[str, str],
+) -> list[dict[str, Any]]:
+    name = spec.name.upper()
+    if name == "DIM_CUSTOMER":
+        return _customer_rows(columns, row_count, rng)
+    if name == "DIM_INVOICE_CATEGORY":
+        return _invoice_category_rows(columns, row_count)
+    if name == "FACT_INVOICES":
+        return _invoice_rows(columns, row_count, rng, table_rows)
+    return _generic_rows(name, columns, row_count, rng, table_rows, primary_keys)
+
+
+def _customer_rows(columns: list[DatasetColumn], row_count: int, rng: random.Random) -> list[dict[str, Any]]:
+    industries = ["LNG", "Data Centers", "Power Generation", "Industrial Gas", "New Energy", "Refining", "Metals & Mining", "Hydrogen", "CCUS"]
+    regions = {
+        "Americas": ["United States", "Canada", "Brazil", "Mexico"],
+        "EMEA": ["United Kingdom", "Germany", "Italy", "Norway"],
+        "APAC": ["India", "Singapore", "Australia", "Japan"],
+        "Middle East": ["Qatar", "United Arab Emirates", "Saudi Arabia", "Oman"],
+    }
+    tiers = ["Strategic", "Growth", "Transactional"]
+    legacy_orgs = ["Baker Hughes", "Chart Industries", "Both"]
+    terms = ["Net 30", "Net 45", "Net 60", "Net 90"]
+    base_names = [
+        "QatarEnergy LNG", "ADNOC Gas Processing", "Saudi Aramco", "NextEra Energy", "Air Liquide",
+        "Linde", "Equinor", "Cheniere Energy", "Rio Tinto", "ArcelorMittal", "Digital Realty", "Air Products",
+    ]
+    rows = []
+    for idx in range(1, row_count + 1):
+        region = rng.choice(list(regions))
+        country = rng.choice(regions[region])
+        industry = rng.choice(industries)
+        legacy = rng.choices(legacy_orgs, weights=[42, 42, 16], k=1)[0]
+        term = rng.choice(terms if legacy != "Baker Hughes" else ["Net 30", "Net 45", "Net 60"])
+        values = {
+            "CUSTOMER_ID": idx,
+            "CUSTOMER_NAME": f"{rng.choice(base_names)} {idx:03d}" if idx > len(base_names) else base_names[idx - 1],
+            "INDUSTRY": industry,
+            "COUNTRY": country,
+            "REGION": region,
+            "CUSTOMER_TIER": rng.choice(tiers),
+            "LEGACY_ORG": legacy,
+            "CONTRACTED_PAYMENT_TERMS": term,
+        }
+        rows.append(_project_row(columns, values, idx, rng))
+    return rows
+
+
+def _invoice_category_rows(columns: list[DatasetColumn], row_count: int) -> list[dict[str, Any]]:
+    categories = [
+        ("New Equipment", "IET", "Project-Based"),
+        ("Aftermarket Parts", "RSL", "Recurring"),
+        ("Field Service", "OFSE", "Recurring"),
+        ("Digital", "IET", "Recurring"),
+        ("Leasing", "Cryo Tank Solutions", "Recurring"),
+        ("Heat Exchanger Overhaul", "Heat Transfer Systems", "Project-Based"),
+        ("Specialty Products", "Specialty Products", "One-Time"),
+    ]
+    rows = []
+    for idx in range(1, row_count + 1):
+        category, business_unit, order_type = categories[(idx - 1) % len(categories)]
+        values = {
+            "CATEGORY_ID": idx,
+            "CATEGORY_NAME": category,
+            "BUSINESS_UNIT": business_unit,
+            "ORDER_TYPE": order_type,
+        }
+        rows.append(_project_row(columns, values, idx, random.Random(idx)))
+    return rows
+
+
+def _invoice_rows(
+    columns: list[DatasetColumn],
+    row_count: int,
+    rng: random.Random,
+    table_rows: dict[str, list[dict[str, Any]]],
+) -> list[dict[str, Any]]:
+    customers = table_rows.get("DIM_CUSTOMER") or [{"CUSTOMER_ID": 1, "REGION": "Americas", "LEGACY_ORG": "Baker Hughes", "CONTRACTED_PAYMENT_TERMS": "Net 45"}]
+    categories = table_rows.get("DIM_INVOICE_CATEGORY") or [{"CATEGORY_ID": 1, "CATEGORY_NAME": "New Equipment", "BUSINESS_UNIT": "IET", "ORDER_TYPE": "Project-Based"}]
+    start = date(2025, 1, 1)
+    end = date(2026, 6, 30)
+    as_of = end
+    dispute_reasons = ["Pricing Discrepancy", "Missing PO", "Quantity Dispute", "Contract Terms"]
+    rows = []
+    for idx in range(1, row_count + 1):
+        customer = rng.choice(customers)
+        category = rng.choice(categories)
+        invoice_date = start + timedelta(days=rng.randint(0, (end - start).days))
+        terms = int(re.search(r"\d+", str(customer.get("CONTRACTED_PAYMENT_TERMS", "Net 45"))).group(0))
+        if customer.get("LEGACY_ORG") == "Chart Industries" and customer.get("REGION") in {"APAC", "Middle East"}:
+            terms = rng.choice([60, 75, 90])
+        due_date = invoice_date + timedelta(days=terms)
+        large_lng_gap = idx <= 15
+        if large_lng_gap:
+            amount = round(rng.uniform(5_000_000, 20_000_000), 2)
+            due_date = as_of - timedelta(days=rng.randint(60, 90))
+            invoice_date = due_date - timedelta(days=terms)
+            paid = False
+            dispute = "N"
+        else:
+            rsl_drag = category.get("BUSINESS_UNIT") == "RSL" and rng.random() < 0.42
+            amount = round(rng.uniform(8_000, 48_000) if rsl_drag else rng.uniform(40_000, 2_500_000), 2)
+            paid = rng.random() < 0.68
+            dispute = "Y" if rng.random() < 0.12 else "N"
+        if paid:
+            pay_delay = max(0, int(rng.gauss(terms + (22 if customer.get("REGION") in {"APAC", "Middle East"} else 4), 18)))
+            payment_date = invoice_date + timedelta(days=pay_delay)
+            if payment_date > as_of:
+                paid = False
+        if paid:
+            payment_received_date = payment_date
+            amount_paid = amount
+            days_outstanding = 0
+        else:
+            payment_received_date = None
+            amount_paid = 0.0
+            days_outstanding = max(0, (as_of - due_date).days)
+        bucket = _aging_bucket(days_outstanding)
+        values = {
+            "INVOICE_ID": idx,
+            "INVOICE_DATE": invoice_date,
+            "DUE_DATE": due_date,
+            "CUSTOMER_ID": customer.get("CUSTOMER_ID"),
+            "CATEGORY_ID": category.get("CATEGORY_ID"),
+            "INVOICE_AMOUNT_USD": amount,
+            "PAYMENT_RECEIVED_DATE": payment_received_date,
+            "AMOUNT_PAID_USD": amount_paid,
+            "DISPUTE_FLAG": dispute,
+            "DISPUTE_REASON": None if dispute == "N" else rng.choice(dispute_reasons),
+            "COLLECTOR_ASSIGNED": "Y" if (not paid and days_outstanding > 30 and rng.random() < 0.72) else "N",
+            "LEGACY_ORG": customer.get("LEGACY_ORG", rng.choice(["Baker Hughes", "Chart Industries"])),
+            "DAYS_OUTSTANDING": days_outstanding,
+            "AGING_BUCKET": bucket,
+            "PAST_DUE_AMOUNT_USD": 0.0 if paid or days_outstanding == 0 else amount,
+        }
+        rows.append(_project_row(columns, values, idx, rng))
+    return rows
+
+
+def _aging_bucket(days: int) -> str:
+    if days <= 0:
+        return "Current"
+    if days <= 30:
+        return "1-30 Days"
+    if days <= 60:
+        return "30-60 Days"
+    if days <= 90:
+        return "60-90 Days"
+    return "90+ Days"
+
+
+def _generic_rows(
+    table_name: str,
+    columns: list[DatasetColumn],
+    row_count: int,
+    rng: random.Random,
+    table_rows: dict[str, list[dict[str, Any]]],
+    primary_keys: dict[str, str],
+) -> list[dict[str, Any]]:
+    rows = []
+    for idx in range(1, row_count + 1):
+        values = {}
+        for column in columns:
+            if column.semantic_role == "fact_key" or column.name == columns[0].name:
+                values[column.name] = idx
+            elif column.name.endswith("_ID"):
+                source_rows = next((rows for name, rows in table_rows.items() if column.name in rows[0]), None)
+                values[column.name] = rng.choice(source_rows)[column.name] if source_rows else idx
+            else:
+                values[column.name] = _generic_value(column.name, idx, rng)
+        rows.append(values)
+    return rows
+
+
+def _project_row(columns: list[DatasetColumn], values: dict[str, Any], idx: int, rng: random.Random) -> dict[str, Any]:
+    return {column.name: values.get(column.name, _generic_value(column.name, idx, rng)) for column in columns}
+
+
+def _generic_value(name: str, idx: int, rng: random.Random) -> Any:
+    if "DATE" in name:
+        return date(2025, 1, 1) + timedelta(days=rng.randint(0, 545))
+    if any(token in name for token in ("AMOUNT", "USD", "REVENUE")):
+        return round(rng.uniform(10_000, 1_000_000), 2)
+    if name.startswith("DAYS_") or name.endswith("_DAYS"):
+        return rng.randint(0, 120)
+    if name.endswith("_ID") or name.endswith("_KEY"):
+        return idx
+    if "FLAG" in name or "ASSIGNED" in name:
+        return rng.choice(["Y", "N"])
+    if "REGION" in name:
+        return rng.choice(["Americas", "EMEA", "APAC", "Middle East"])
+    if "COUNTRY" in name:
+        return rng.choice(["United States", "Qatar", "United Arab Emirates", "Germany", "India"])
+    if "LEGACY_ORG" in name:
+        return rng.choice(["Baker Hughes", "Chart Industries", "Both"])
+    return f"{name.title().replace('_', ' ')} {idx:03d}"

demoprep_app/dataset/generators/post_merger_integration.py

@@ -1,610 +0,0 @@
-"""Post-merger integration dataset generator.
-
-This generator is designed for acquisition synergy demos where the story spans
-procurement cost synergy, aftermarket revenue acceleration, and cash-flow/AR
-improvement. It is intentionally scenario-family based rather than tied to a
-single company name.
-"""
-
-from __future__ import annotations
-
-import random
-from datetime import date, timedelta
-
-from demoprep_app.dataset.contracts import DatasetBundle, DatasetColumn, DatasetTable
-from demoprep_app.dataset.generators.base import ScenarioDatasetGenerator
-from demoprep_app.scenario.contract import ScenarioContract
-
-
-class PostMergerIntegrationDatasetGenerator(ScenarioDatasetGenerator):
-    scenario_types = ("post_merger_integration", "acquisition_synergy")
-
-    def generate(self, scenario: ScenarioContract, row_count: int | None = None) -> DatasetBundle:
-        rng = random.Random(scenario.metadata.get("seed", 202606))
-        scale = min(1.0, max(0.02, (row_count or 10000) / 10000.0))
-
-        suppliers = self._suppliers(rng)
-        commodities = self._commodities()
-        customers = self._customers(rng)
-        products = self._products()
-        invoice_categories = self._invoice_categories()
-
-        spend_rows = self._spend_rows(rng, suppliers, commodities, int(50000 * min(scale, 1.0)))
-        installed_rows = self._installed_rows(rng, customers, products, int(5000 * min(scale, 1.0)))
-        order_rows = self._order_rows(rng, customers, products, installed_rows, int(30000 * min(scale, 1.0)))
-        invoice_rows = self._invoice_rows(rng, customers, invoice_categories, int(25000 * min(scale, 1.0)))
-
-        return DatasetBundle(
-            scenario=scenario,
-            tables=[
-                DatasetTable("DIM_SUPPLIER", "one row per supplier", self._supplier_columns(), suppliers),
-                DatasetTable("DIM_COMMODITY", "one row per commodity", self._commodity_columns(), commodities),
-                DatasetTable("DIM_CUSTOMER", "one row per customer", self._customer_columns(), customers),
-                DatasetTable("DIM_PRODUCT", "one row per product", self._product_columns(), products),
-                DatasetTable("DIM_INVOICE_CATEGORY", "one row per invoice category", self._invoice_category_columns(), invoice_categories),
-                DatasetTable("FACT_SPEND", "purchase-order supplier commodity site day", self._spend_columns(), spend_rows, is_fact=True),
-                DatasetTable("FACT_EQUIPMENT_INSTALLED", "installed equipment unit", self._installed_columns(), installed_rows, is_fact=True),
-                DatasetTable("FACT_ORDERS", "customer product order day", self._order_columns(), order_rows, is_fact=True),
-                DatasetTable("FACT_INVOICES", "customer invoice day", self._invoice_columns(), invoice_rows, is_fact=True),
-            ],
-        )
-
-    @staticmethod
-    def _supplier_columns() -> list[DatasetColumn]:
-        return [
-            DatasetColumn("SUPPLIER_ID", "supplier_key", "NUMBER", nullable=False),
-            DatasetColumn("SUPPLIER_NAME", "supplier_name", "VARCHAR(120)", nullable=False),
-            DatasetColumn("SUPPLIER_CATEGORY", "category", "VARCHAR(80)", nullable=False),
-            DatasetColumn("COUNTRY", "country", "VARCHAR(80)", nullable=False),
-            DatasetColumn("REGION", "region", "VARCHAR(40)", nullable=False),
-            DatasetColumn("STRATEGIC_TIER", "tier", "VARCHAR(40)", nullable=False),
-            DatasetColumn("LEGACY_ORG", "legacy_org", "VARCHAR(40)", nullable=False),
-            DatasetColumn("IS_DUPLICATE_SUPPLIER", "flag", "BOOLEAN", nullable=False),
-        ]
-
-    @staticmethod
-    def _commodity_columns() -> list[DatasetColumn]:
-        return [
-            DatasetColumn("COMMODITY_CODE", "commodity_key", "NUMBER", nullable=False),
-            DatasetColumn("COMMODITY_NAME", "commodity_name", "VARCHAR(120)", nullable=False),
-            DatasetColumn("COMMODITY_GROUP", "commodity_group", "VARCHAR(80)", nullable=False),
-        ]
-
-    @staticmethod
-    def _customer_columns() -> list[DatasetColumn]:
-        return [
-            DatasetColumn("CUSTOMER_ID", "customer_key", "NUMBER", nullable=False),
-            DatasetColumn("CUSTOMER_NAME", "customer_name", "VARCHAR(120)", nullable=False),
-            DatasetColumn("INDUSTRY", "industry", "VARCHAR(80)", nullable=False),
-            DatasetColumn("COUNTRY", "country", "VARCHAR(80)", nullable=False),
-            DatasetColumn("REGION", "region", "VARCHAR(40)", nullable=False),
-            DatasetColumn("CUSTOMER_TIER", "tier", "VARCHAR(40)", nullable=False),
-            DatasetColumn("LEGACY_ORG", "legacy_org", "VARCHAR(40)", nullable=False),
-            DatasetColumn("CONTRACTED_PAYMENT_TERMS", "payment_terms", "VARCHAR(20)", nullable=False),
-        ]
-
-    @staticmethod
-    def _product_columns() -> list[DatasetColumn]:
-        return [
-            DatasetColumn("PRODUCT_ID", "product_key", "NUMBER", nullable=False),
-            DatasetColumn("PRODUCT_NAME", "product_name", "VARCHAR(120)", nullable=False),
-            DatasetColumn("PRODUCT_CATEGORY", "category", "VARCHAR(80)", nullable=False),
-            DatasetColumn("LEGACY_ORG", "legacy_org", "VARCHAR(40)", nullable=False),
-            DatasetColumn("BUSINESS_UNIT", "business_unit", "VARCHAR(80)", nullable=False),
-            DatasetColumn("SERVICEABLE", "flag", "BOOLEAN", nullable=False),
-        ]
-
-    @staticmethod
-    def _invoice_category_columns() -> list[DatasetColumn]:
-        return [
-            DatasetColumn("CATEGORY_ID", "category_key", "NUMBER", nullable=False),
-            DatasetColumn("CATEGORY_NAME", "category_name", "VARCHAR(80)", nullable=False),
-            DatasetColumn("BUSINESS_UNIT", "business_unit", "VARCHAR(80)", nullable=False),
-            DatasetColumn("ORDER_TYPE", "order_type", "VARCHAR(40)", nullable=False),
-        ]
-
-    @staticmethod
-    def _spend_columns() -> list[DatasetColumn]:
-        return [
-            DatasetColumn("SPEND_ID", "fact_key", "NUMBER", nullable=False),
-            DatasetColumn("PO_ID", "po_id", "VARCHAR(30)", nullable=False),
-            DatasetColumn("PO_DATE", "date", "DATE", nullable=False),
-            DatasetColumn("SUPPLIER_ID", "supplier_key", "NUMBER", nullable=False),
-            DatasetColumn("COMMODITY_CODE", "commodity_key", "NUMBER", nullable=False),
-            DatasetColumn("SPEND_AMOUNT_USD", "spend", "NUMBER(14,2)", nullable=False),
-            DatasetColumn("CONTRACTED_PAYMENT_TERMS", "payment_terms", "VARCHAR(20)", nullable=False),
-            DatasetColumn("INVOICE_DATE", "date", "DATE", nullable=False),
-            DatasetColumn("PAYMENT_DATE", "date", "DATE", nullable=False),
-            DatasetColumn("SITE_NAME", "site", "VARCHAR(100)", nullable=False),
-            DatasetColumn("SITE_CITY", "city", "VARCHAR(80)", nullable=False),
-            DatasetColumn("SITE_COUNTRY", "country", "VARCHAR(80)", nullable=False),
-            DatasetColumn("SITE_REGION", "region", "VARCHAR(40)", nullable=False),
-            DatasetColumn("SITE_TYPE", "site_type", "VARCHAR(40)", nullable=False),
-            DatasetColumn("SITE_LEGACY_ORG", "legacy_org", "VARCHAR(40)", nullable=False),
-            DatasetColumn("BUSINESS_UNIT", "business_unit", "VARCHAR(80)", nullable=False),
-            DatasetColumn("CONSOLIDATION_OPPORTUNITY_USD", "opportunity", "NUMBER(14,2)", nullable=False),
-        ]
-
-    @staticmethod
-    def _installed_columns() -> list[DatasetColumn]:
-        return [
-            DatasetColumn("INSTALLATION_ID", "fact_key", "NUMBER", nullable=False),
-            DatasetColumn("CUSTOMER_ID", "customer_key", "NUMBER", nullable=False),
-            DatasetColumn("PRODUCT_ID", "product_key", "NUMBER", nullable=False),
-            DatasetColumn("SITE_COUNTRY", "country", "VARCHAR(80)", nullable=False),
-            DatasetColumn("SITE_REGION", "region", "VARCHAR(40)", nullable=False),
-            DatasetColumn("INSTALLATION_DATE", "date", "DATE", nullable=False),
-            DatasetColumn("EXPECTED_OVERHAUL_DATE", "date", "DATE", nullable=False),
-            DatasetColumn("SERVICE_CONTRACT_IN_PLACE", "flag", "VARCHAR(1)", nullable=False),
-            DatasetColumn("SERVICE_CONTRACT_TYPE", "contract_type", "VARCHAR(40)", nullable=False),
-            DatasetColumn("LAST_SERVICE_DATE", "date", "DATE", nullable=True),
-            DatasetColumn("ASSIGNED_SERVICE_ORG", "service_org", "VARCHAR(40)", nullable=False),
-            DatasetColumn("AFTERMARKET_GAP_USD", "opportunity", "NUMBER(14,2)", nullable=False),
-        ]
-
-    @staticmethod
-    def _order_columns() -> list[DatasetColumn]:
-        return [
-            DatasetColumn("ORDER_ID", "fact_key", "NUMBER", nullable=False),
-            DatasetColumn("ORDER_DATE", "date", "DATE", nullable=False),
-            DatasetColumn("CUSTOMER_ID", "customer_key", "NUMBER", nullable=False),
-            DatasetColumn("PRODUCT_ID", "product_key", "NUMBER", nullable=False),
-            DatasetColumn("ORDER_TYPE", "order_type", "VARCHAR(40)", nullable=False),
-            DatasetColumn("REVENUE_USD", "revenue", "NUMBER(14,2)", nullable=False),
-            DatasetColumn("MARGIN_USD", "margin", "NUMBER(14,2)", nullable=False),
-            DatasetColumn("PROJECT_ID", "project", "VARCHAR(30)", nullable=True),
-            DatasetColumn("LEGACY_ORG_FULFILLING_ORDER", "legacy_org", "VARCHAR(40)", nullable=False),
-            DatasetColumn("CROSS_SELL_GAP_FLAG", "flag", "BOOLEAN", nullable=False),
-        ]
-
-    @staticmethod
-    def _invoice_columns() -> list[DatasetColumn]:
-        return [
-            DatasetColumn("INVOICE_ID", "fact_key", "NUMBER", nullable=False),
-            DatasetColumn("INVOICE_DATE", "date", "DATE", nullable=False),
-            DatasetColumn("DUE_DATE", "date", "DATE", nullable=False),
-            DatasetColumn("CUSTOMER_ID", "customer_key", "NUMBER", nullable=False),
-            DatasetColumn("CATEGORY_ID", "category_key", "NUMBER", nullable=False),
-            DatasetColumn("INVOICE_AMOUNT_USD", "amount", "NUMBER(14,2)", nullable=False),
-            DatasetColumn("PAYMENT_RECEIVED_DATE", "date", "DATE", nullable=True),
-            DatasetColumn("AMOUNT_PAID_USD", "amount", "NUMBER(14,2)", nullable=False),
-            DatasetColumn("DAYS_OUTSTANDING", "days", "NUMBER", nullable=False),
-            DatasetColumn("AGING_BUCKET", "aging_bucket", "VARCHAR(30)", nullable=False),
-            DatasetColumn("DISPUTE_FLAG", "flag", "VARCHAR(1)", nullable=False),
-            DatasetColumn("DISPUTE_REASON", "reason", "VARCHAR(80)", nullable=True),
-            DatasetColumn("COLLECTOR_ASSIGNED", "flag", "VARCHAR(1)", nullable=False),
-            DatasetColumn("LEGACY_ORG", "legacy_org", "VARCHAR(40)", nullable=False),
-            DatasetColumn("CASH_ACCELERATION_OPPORTUNITY_USD", "opportunity", "NUMBER(14,2)", nullable=False),
-        ]
-
-    def _suppliers(self, rng: random.Random) -> list[dict]:
-        supplier_names = [
-            "Allegheny Specialty Metals", "Ametek Process Instruments", "Atlas Copco Compressors",
-            "BASF Industrial Coatings", "Bekaert Steel Wire", "BorgWarner Thermal Systems",
-            "Bray International", "Chartwell Cryogenic Services", "Curtiss-Wright Valves",
-            "Dover Precision Components", "Emerson Automation Solutions", "Flowserve Corporation",
-            "Gardner Denver Nash", "Hubbell Industrial Controls", "ITT Engineered Valves",
-            "John Crane Sealing Systems", "Linde Engineering Services", "Marmon Industrial Energy",
-            "Mitsubishi Heavy Compressor", "Mueller Cryogenic Products", "Neles Flow Control",
-            "Parker Hannifin Motion", "Rotork Controls", "Sandvik Materials Technology",
-            "Schneider Electric", "Siemens Energy Controls", "SKF Industrial Bearings",
-            "Sulzer Turbo Services", "Technip Energies Logistics", "Tenaris Precision Tubes",
-            "Thermo Fisher Controls", "Timken Power Systems", "Trane Industrial Fans",
-            "Valmet Automation", "Viega Industrial Piping", "Wartsila Gas Solutions",
-            "Woodward Turbine Controls", "Xylem Cryogenic Pumps", "Yokogawa Process Systems",
-            "Zeeco Combustion Controls",
-        ]
-        countries = [
-            ("United States", "Americas"), ("Mexico", "Americas"), ("Germany", "EMEA"),
-            ("Italy", "EMEA"), ("United Kingdom", "EMEA"), ("India", "APAC"),
-            ("China", "APAC"), ("Japan", "APAC"), ("Singapore", "APAC"),
-            ("United Arab Emirates", "Middle East"),
-        ]
-        categories = ["Raw Materials", "Engineered Components", "MRO", "Logistics", "IT & Software", "Professional Services"]
-        rows = []
-        supplier_id = 1
-        duplicate_base = supplier_names[:40] if len(supplier_names) >= 40 else supplier_names
-        for name in duplicate_base:
-            country, region = rng.choice(countries)
-            rows.append({
-                "SUPPLIER_ID": supplier_id,
-                "SUPPLIER_NAME": name,
-                "SUPPLIER_CATEGORY": "Engineered Components" if supplier_id <= 30 else "IT & Software",
-                "COUNTRY": country,
-                "REGION": region,
-                "STRATEGIC_TIER": "Approved",
-                "LEGACY_ORG": "Both",
-                "IS_DUPLICATE_SUPPLIER": True,
-            })
-            supplier_id += 1
-        while supplier_id <= 300:
-            category = rng.choices(categories, weights=[18, 26, 18, 12, 10, 6], k=1)[0]
-            country, region = rng.choice(countries)
-            legacy = rng.choices(["Baker Hughes", "Chart Industries"], weights=[58, 42], k=1)[0]
-            tail = legacy == "Chart Industries" and region == "APAC" and supplier_id <= 240
-            rows.append({
-                "SUPPLIER_ID": supplier_id,
-                "SUPPLIER_NAME": self._supplier_name(category, supplier_id, rng),
-                "SUPPLIER_CATEGORY": category,
-                "COUNTRY": country,
-                "REGION": region,
-                "STRATEGIC_TIER": "Tail" if tail else rng.choices(["Preferred", "Approved", "Tail"], weights=[18, 58, 24], k=1)[0],
-                "LEGACY_ORG": legacy,
-                "IS_DUPLICATE_SUPPLIER": False,
-            })
-            supplier_id += 1
-        return rows
-
-    @staticmethod
-    def _supplier_name(category: str, supplier_id: int, rng: random.Random) -> str:
-        prefixes = {
-            "Raw Materials": ["Summit Alloy", "Frontier Steel", "Crescent Aluminum", "Keystone Metals", "Allegheny Nickel", "Blue Ridge Metals"],
-            "Engineered Components": ["Precision Turbine", "CryoValve Systems", "Apex Compressor", "ThermalCore", "Vector Flow", "Atlas Controls"],
-            "MRO": ["PlantCare Industrial", "Reliant MRO", "MaintainPro Services", "Northstar Repair", "Sagefield Maintenance", "Evergreen Industrial"],
-            "Logistics": ["GlobalLift Logistics", "HarborLine Freight", "Vector Heavy Haul", "TransOcean Forwarding", "Bridgeport Logistics", "Meridian Cargo"],
-            "IT & Software": ["CloudBridge Systems", "SecureWorks Industrial", "ERPWorks", "DataLink Software", "Northstar Cloud", "Helix Applications"],
-            "Professional Services": ["Pinnacle Advisory", "Integration Partners", "Apex Consulting", "NorthBridge Legal", "Cedar Strategy", "HarborPoint Advisors"],
-        }
-        suffixes = ["Group", "Solutions", "Services", "Industries", "Partners", "Supply", "Works", "Technologies", "Associates", "International"]
-        regions = ["Americas", "Gulf Coast", "North Sea", "Adriatic", "Singapore", "Midlands", "Bavaria", "Piedmont", "Kanto", "Jebel Ali"]
-        prefix = prefixes[category][supplier_id % len(prefixes[category])]
-        suffix = suffixes[(supplier_id // len(prefixes[category])) % len(suffixes)]
-        region = regions[(supplier_id // 7) % len(regions)]
-        return f"{region} {prefix} {suffix}"
-
-    @staticmethod
-    def _commodities() -> list[dict]:
-        items = [
-            ("Aluminum Alloys", "Raw Materials"), ("Steel Forgings", "Raw Materials"),
-            ("Nickel Alloy Tubing", "Raw Materials"), ("Stainless Plate", "Raw Materials"),
-            ("Brazed Aluminum Heat Exchanger Cores", "Engineered Components"),
-            ("Centrifugal Compressor Impellers", "Engineered Components"),
-            ("Cryogenic Valves", "Engineered Components"), ("Control Valves", "Engineered Components"),
-            ("Industrial Fans and Blowers", "Engineered Components"), ("Electronic Controls", "Engineered Components"),
-            ("Insulation Materials", "Raw Materials"), ("Rotating Equipment Bearings", "Engineered Components"),
-            ("Seals and Gaskets", "MRO"), ("Field Service Tooling", "MRO"),
-            ("Freight and Heavy Haul", "Logistics"), ("Ocean Freight", "Logistics"),
-            ("ERP Software Licenses", "IT & Software"), ("Engineering CAD Subscriptions", "IT & Software"),
-            ("Cybersecurity Subscriptions", "IT & Software"), ("Cloud Data Platforms", "IT & Software"),
-            ("Copper Bus Bar Assemblies", "Engineered Components"), ("Pressure Vessel Shells", "Engineered Components"),
-            ("Plate Fin Assemblies", "Engineered Components"), ("Compressor Dry Gas Seals", "Engineered Components"),
-            ("Turbine Blade Coatings", "Engineered Components"), ("Cryogenic Pump Skids", "Engineered Components"),
-            ("Thermal Insulation Jackets", "Raw Materials"), ("Specialty Welding Consumables", "MRO"),
-            ("Instrumentation Calibration", "MRO"), ("Plant Safety Supplies", "MRO"),
-            ("Air Freight Expedites", "Logistics"), ("Regional Trucking Lanes", "Logistics"),
-            ("Customs Brokerage", "Logistics"), ("Warehouse Handling", "Logistics"),
-            ("PLM Software Seats", "IT & Software"), ("Procurement Platform Licenses", "IT & Software"),
-            ("Manufacturing Execution Systems", "IT & Software"), ("Managed Network Services", "IT & Software"),
-            ("Integration Advisory Services", "Professional Services"), ("Engineering Contractor Services", "Professional Services"),
-        ]
-        return [
-            {"COMMODITY_CODE": idx, "COMMODITY_NAME": name, "COMMODITY_GROUP": group}
-            for idx, (name, group) in enumerate(items, start=1)
-        ]
-
-    def _customers(self, rng: random.Random) -> list[dict]:
-        base = [
-            ("QatarEnergy LNG", "LNG", "Qatar", "Middle East", "Strategic"),
-            ("ADNOC Gas Processing", "LNG", "United Arab Emirates", "Middle East", "Strategic"),
-            ("Saudi Aramco", "Refining", "Saudi Arabia", "Middle East", "Strategic"),
-            ("Microsoft Data Center Operations", "Data Centers", "United States", "Americas", "Strategic"),
-            ("Amazon Web Services", "Data Centers", "United States", "Americas", "Strategic"),
-            ("Air Liquide", "Industrial Gas", "France", "EMEA", "Strategic"),
-            ("Linde", "Industrial Gas", "Germany", "EMEA", "Strategic"),
-            ("JERA Power", "Power Generation", "Japan", "APAC", "Strategic"),
-            ("Rio Tinto Minerals", "Metals & Mining", "Australia", "APAC", "Growth"),
-            ("Plug Power Hydrogen", "New Energy", "United States", "Americas", "Growth"),
-            ("Occidental Low Carbon Ventures", "New Energy", "United States", "Americas", "Growth"),
-        ]
-        industries = ["LNG", "Data Centers", "Power Generation", "Industrial Gas", "New Energy", "Refining", "Metals & Mining"]
-        countries = [
-            ("United States", "Americas"), ("Brazil", "Americas"), ("Germany", "EMEA"), ("France", "EMEA"),
-            ("United Kingdom", "EMEA"), ("India", "APAC"), ("Japan", "APAC"), ("Australia", "APAC"),
-            ("Qatar", "Middle East"), ("United Arab Emirates", "Middle East"), ("Saudi Arabia", "Middle East"),
-        ]
-        rows = []
-        for idx, (name, industry, country, region, tier) in enumerate(base, start=1):
-            rows.append(self._customer_row(idx, name, industry, country, region, tier, rng))
-        for idx in range(len(rows) + 1, 201):
-            industry = rng.choice(industries)
-            country, region = rng.choice(countries)
-            tier = rng.choices(["Strategic", "Growth", "Transactional"], weights=[18, 44, 38], k=1)[0]
-            name = f"{rng.choice(['Atlas', 'Bluewater', 'Cedar', 'Frontier', 'Helios', 'Nexus', 'Orion', 'Summit'])} {industry} {rng.choice(['Holdings', 'Operations', 'Energy', 'Systems', 'Partners'])}"
-            rows.append(self._customer_row(idx, name, industry, country, region, tier, rng))
-        return rows
-
-    @staticmethod
-    def _customer_row(idx: int, name: str, industry: str, country: str, region: str, tier: str, rng: random.Random) -> dict:
-        legacy = rng.choices(["Baker Hughes", "Chart Industries", "Both"], weights=[48, 36, 16], k=1)[0]
-        terms = rng.choices(["Net 30", "Net 45", "Net 60", "Net 90"], weights=[36, 28, 24, 12], k=1)[0]
-        if legacy == "Chart Industries" and region in {"APAC", "Middle East"}:
-            terms = rng.choices(["Net 60", "Net 90"], weights=[45, 55], k=1)[0]
-        return {
-            "CUSTOMER_ID": idx,
-            "CUSTOMER_NAME": name,
-            "INDUSTRY": industry,
-            "COUNTRY": country,
-            "REGION": region,
-            "CUSTOMER_TIER": tier,
-            "LEGACY_ORG": legacy,
-            "CONTRACTED_PAYMENT_TERMS": terms,
-        }
-
-    @staticmethod
-    def _products() -> list[dict]:
-        product_defs = [
-            ("LM9000 Gas Turbine Package", "Rotating Equipment", "Baker Hughes", "IET", True),
-            ("NovaLT Compressor Train", "Rotating Equipment", "Baker Hughes", "IET", True),
-            ("Centrifugal Compressor Bundle", "Rotating Equipment", "Baker Hughes", "IET", True),
-            ("Masoneilan Control Valve", "Flow Control", "Baker Hughes", "IET", True),
-            ("Bently Nevada Monitoring Suite", "Digital & Services", "Baker Hughes", "IET", False),
-            ("Brazed Aluminum Heat Exchanger", "Heat Transfer", "Chart Industries", "Heat Transfer Systems", True),
-            ("Cryogenic Storage Tank", "Cryogenic Equipment", "Chart Industries", "Cryo Tank Solutions", True),
-            ("Hydrogen Liquefaction Cold Box", "Cryogenic Equipment", "Chart Industries", "Specialty Products", True),
-            ("Hudson Industrial Fan", "Industrial Fans & Blowers", "Chart Industries", "RSL", True),
-            ("Howden Roots Blower", "Industrial Fans & Blowers", "Chart Industries", "RSL", True),
-        ]
-        variants = [
-            "Base", "High Efficiency", "Low Emissions", "Arctic Service", "Desert Service",
-            "Modular", "Compact", "High Pressure", "Low Temperature", "Digital Ready",
-            "Service Plus", "Reliability", "Premium", "Extended Duty", "Rapid Deploy",
-        ]
-        rows = []
-        for idx in range(1, 151):
-            name, category, legacy, bu, serviceable = product_defs[(idx - 1) % len(product_defs)]
-            suffix = "" if idx <= len(product_defs) else f" - {variants[((idx - 1) // len(product_defs) - 1) % len(variants)]}"
-            rows.append({
-                "PRODUCT_ID": idx,
-                "PRODUCT_NAME": f"{name}{suffix}",
-                "PRODUCT_CATEGORY": category,
-                "LEGACY_ORG": legacy,
-                "BUSINESS_UNIT": bu,
-                "SERVICEABLE": serviceable,
-            })
-        return rows
-
-    @staticmethod
-    def _invoice_categories() -> list[dict]:
-        categories = [
-            ("New Equipment", "IET", "Project-Based"), ("Aftermarket Parts", "IET", "Recurring"),
-            ("Field Service", "IET", "Recurring"), ("Digital", "IET", "Recurring"),
-            ("Leasing", "Cryo Tank Solutions", "Recurring"), ("New Equipment", "Cryo Tank Solutions", "Project-Based"),
-            ("Aftermarket Parts", "Heat Transfer Systems", "Recurring"), ("Field Service", "RSL", "One-Time"),
-            ("Digital", "RSL", "Recurring"), ("New Equipment", "Specialty Products", "Project-Based"),
-        ]
-        while len(categories) < 20:
-            categories.append(categories[len(categories) % 10])
-        return [
-            {"CATEGORY_ID": idx, "CATEGORY_NAME": name, "BUSINESS_UNIT": bu, "ORDER_TYPE": order_type}
-            for idx, (name, bu, order_type) in enumerate(categories, start=1)
-        ]
-
-    def _spend_rows(self, rng: random.Random, suppliers: list[dict], commodities: list[dict], count: int) -> list[dict]:
-        start = date(2025, 1, 1)
-        days = (date(2026, 6, 30) - start).days
-        sites = [
-            ("Florence Turbomachinery Plant", "Florence", "Italy", "EMEA", "Manufacturing", "Baker Hughes", "IET"),
-            ("Houston IET Service Center", "Houston", "United States", "Americas", "Service Center", "Baker Hughes", "IET"),
-            ("Tulsa Compressor Works", "Tulsa", "United States", "Americas", "Manufacturing", "Baker Hughes", "OFSE"),
-            ("The Woodlands Integration Office", "The Woodlands", "United States", "Americas", "Office", "Baker Hughes", "IET"),
-            ("Ball Ground Cryogenic Plant", "Ball Ground", "United States", "Americas", "Manufacturing", "Chart Industries", "Cryo Tank Solutions"),
-            ("New Prague Heat Transfer Plant", "New Prague", "United States", "Americas", "Manufacturing", "Chart Industries", "Heat Transfer Systems"),
-            ("Changzhou Cryogenic Plant", "Changzhou", "China", "APAC", "Manufacturing", "Chart Industries", "Cryo Tank Solutions"),
-            ("Singapore RSL Service Hub", "Singapore", "Singapore", "APAC", "Service Center", "Chart Industries", "RSL"),
-        ]
-        dup_suppliers = [s for s in suppliers if s["IS_DUPLICATE_SUPPLIER"]]
-        chart_tail_apac = [s for s in suppliers if s["LEGACY_ORG"] == "Chart Industries" and s["REGION"] == "APAC" and s["STRATEGIC_TIER"] == "Tail"]
-        it_commodities = [c for c in commodities if c["COMMODITY_GROUP"] == "IT & Software"]
-        engineered = [c for c in commodities if c["COMMODITY_GROUP"] == "Engineered Components"]
-        rows = []
-        for idx in range(1, count + 1):
-            anomaly_roll = rng.random()
-            if anomaly_roll < 0.26:
-                supplier = rng.choice(dup_suppliers)
-                commodity = rng.choice(engineered)
-                amount = rng.uniform(75000, 650000)
-            elif anomaly_roll < 0.44 and chart_tail_apac:
-                supplier = rng.choice(chart_tail_apac)
-                commodity = rng.choice(commodities)
-                amount = rng.uniform(3500, 22000)
-            elif anomaly_roll < 0.54:
-                supplier = rng.choice([s for s in suppliers if s["SUPPLIER_CATEGORY"] == "IT & Software"])
-                commodity = rng.choice(it_commodities)
-                amount = rng.uniform(18000, 175000)
-            else:
-                supplier = rng.choice(suppliers)
-                commodity = rng.choice(commodities)
-                amount = rng.uniform(8000, 260000)
-            site = rng.choice(sites)
-            po_date = start + timedelta(days=rng.randint(0, days))
-            invoice_date = po_date + timedelta(days=rng.randint(1, 12))
-            if site[5] == "Baker Hughes" and commodity["COMMODITY_NAME"] in {"Aluminum Alloys", "Steel Forgings"}:
-                terms = "Net 30"
-            elif site[5] == "Chart Industries" and commodity["COMMODITY_NAME"] in {"Aluminum Alloys", "Steel Forgings"}:
-                terms = rng.choice(["Net 60", "Net 90"])
-            else:
-                terms = rng.choice(["Net 30", "Net 45", "Net 60", "Net 90"])
-            term_days = int(terms.split()[1])
-            payment_date = invoice_date + timedelta(days=max(5, int(rng.gauss(term_days, 8))))
-            consolidation_opportunity = amount * (0.10 if supplier["IS_DUPLICATE_SUPPLIER"] else 0.04 if supplier["SUPPLIER_CATEGORY"] == "IT & Software" else 0.0)
-            rows.append({
-                "SPEND_ID": idx,
-                "PO_ID": f"PO-{po_date.year}-{idx:07d}",
-                "PO_DATE": po_date.isoformat(),
-                "SUPPLIER_ID": supplier["SUPPLIER_ID"],
-                "COMMODITY_CODE": commodity["COMMODITY_CODE"],
-                "SPEND_AMOUNT_USD": round(amount, 2),
-                "CONTRACTED_PAYMENT_TERMS": terms,
-                "INVOICE_DATE": invoice_date.isoformat(),
-                "PAYMENT_DATE": payment_date.isoformat(),
-                "SITE_NAME": site[0],
-                "SITE_CITY": site[1],
-                "SITE_COUNTRY": site[2],
-                "SITE_REGION": site[3],
-                "SITE_TYPE": site[4],
-                "SITE_LEGACY_ORG": site[5],
-                "BUSINESS_UNIT": site[6],
-                "CONSOLIDATION_OPPORTUNITY_USD": round(consolidation_opportunity, 2),
-            })
-        return rows
-
-    def _installed_rows(self, rng: random.Random, customers: list[dict], products: list[dict], count: int) -> list[dict]:
-        rows = []
-        start = date(2016, 1, 1)
-        for idx in range(1, count + 1):
-            customer = rng.choice(customers)
-            product = rng.choice([p for p in products if p["SERVICEABLE"]])
-            install_date = start + timedelta(days=rng.randint(0, 3650))
-            expected_overhaul = install_date + timedelta(days=rng.randint(1800, 2900))
-            chart_gap = product["LEGACY_ORG"] == "Chart Industries" and customer["REGION"] in {"APAC", "Middle East"}
-            near_overhaul = product["PRODUCT_CATEGORY"] == "Heat Transfer" and date(2018, 1, 1) <= install_date <= date(2021, 12, 31)
-            attach_prob = 0.11 if chart_gap else 0.58 if product["LEGACY_ORG"] == "Baker Hughes" else 0.28
-            has_contract = rng.random() < attach_prob and not near_overhaul
-            contract_type = rng.choice(["Full", "Preventive"]) if has_contract else "None"
-            last_service = install_date + timedelta(days=rng.randint(120, max(121, (date(2026, 6, 30) - install_date).days)))
-            aftermarket_gap = 0.0 if has_contract else rng.uniform(85000, 750000)
-            rows.append({
-                "INSTALLATION_ID": idx,
-                "CUSTOMER_ID": customer["CUSTOMER_ID"],
-                "PRODUCT_ID": product["PRODUCT_ID"],
-                "SITE_COUNTRY": customer["COUNTRY"],
-                "SITE_REGION": customer["REGION"],
-                "INSTALLATION_DATE": install_date.isoformat(),
-                "EXPECTED_OVERHAUL_DATE": expected_overhaul.isoformat(),
-                "SERVICE_CONTRACT_IN_PLACE": "Y" if has_contract else "N",
-                "SERVICE_CONTRACT_TYPE": contract_type,
-                "LAST_SERVICE_DATE": last_service.isoformat() if has_contract or rng.random() < 0.35 else None,
-                "ASSIGNED_SERVICE_ORG": rng.choice(["Baker Hughes", "Chart RSL", "Third Party"]) if has_contract else rng.choice(["Third Party", "None"]),
-                "AFTERMARKET_GAP_USD": round(aftermarket_gap, 2),
-            })
-        return rows
-
-    def _order_rows(self, rng: random.Random, customers: list[dict], products: list[dict], installed: list[dict], count: int) -> list[dict]:
-        start = date(2024, 1, 1)
-        days = (date(2026, 6, 30) - start).days
-        project_ids = [f"PRJ-LNG-{idx:03d}" for idx in range(1, 19)] + [f"PRJ-DC-{idx:03d}" for idx in range(1, 13)]
-        project_customers = {pid: rng.choice([c for c in customers if c["INDUSTRY"] in {"LNG", "Data Centers"}]) for pid in project_ids}
-        rows = []
-        for idx in range(1, count + 1):
-            order_date = start + timedelta(days=rng.randint(0, days))
-            if idx <= len(project_ids) * 2:
-                project_id = project_ids[(idx - 1) // 2]
-                customer = project_customers[project_id]
-                product = rng.choice([p for p in products if p["LEGACY_ORG"] == ("Baker Hughes" if idx % 2 else "Chart Industries")])
-                order_type = "New Equipment"
-                cross_gap = product["LEGACY_ORG"] == "Chart Industries"
-            else:
-                customer = rng.choice(customers)
-                product = rng.choice(products)
-                project_id = None if rng.random() < 0.82 else rng.choice(project_ids)
-                if customer["INDUSTRY"] in {"New Energy"} and product["LEGACY_ORG"] == "Chart Industries":
-                    order_type = rng.choices(["New Equipment", "Aftermarket Parts", "Service", "Upgrade", "Digital"], weights=[82, 4, 3, 8, 3], k=1)[0]
-                else:
-                    order_type = rng.choices(["New Equipment", "Aftermarket Parts", "Service", "Upgrade", "Digital"], weights=[30, 28, 24, 12, 6], k=1)[0]
-                cross_gap = bool(project_id and product["LEGACY_ORG"] == "Chart Industries" and order_type == "New Equipment")
-            base = {
-                "New Equipment": rng.uniform(250000, 8_500_000),
-                "Aftermarket Parts": rng.uniform(12000, 425000),
-                "Service": rng.uniform(25000, 850000),
-                "Upgrade": rng.uniform(90000, 1_600_000),
-                "Digital": rng.uniform(15000, 280000),
-            }[order_type]
-            margin_pct = {"New Equipment": 0.22, "Aftermarket Parts": 0.36, "Service": 0.42, "Upgrade": 0.34, "Digital": 0.48}[order_type]
-            revenue = round(base * rng.uniform(0.72, 1.36), 2)
-            margin = round(revenue * rng.uniform(margin_pct - 0.06, margin_pct + 0.08), 2)
-            rows.append({
-                "ORDER_ID": idx,
-                "ORDER_DATE": order_date.isoformat(),
-                "CUSTOMER_ID": customer["CUSTOMER_ID"],
-                "PRODUCT_ID": product["PRODUCT_ID"],
-                "ORDER_TYPE": order_type,
-                "REVENUE_USD": revenue,
-                "MARGIN_USD": margin,
-                "PROJECT_ID": project_id,
-                "LEGACY_ORG_FULFILLING_ORDER": product["LEGACY_ORG"],
-                "CROSS_SELL_GAP_FLAG": cross_gap,
-            })
-        return rows
-
-    def _invoice_rows(self, rng: random.Random, customers: list[dict], categories: list[dict], count: int) -> list[dict]:
-        start = date(2025, 1, 1)
-        end = date(2026, 6, 30)
-        days = (end - start).days
-        rows = []
-        lng_customers = [c for c in customers if c["INDUSTRY"] == "LNG"]
-        chart_apac_me = [c for c in customers if c["LEGACY_ORG"] in {"Chart Industries", "Both"} and c["REGION"] in {"APAC", "Middle East"}]
-        rsl_categories = [c for c in categories if c["BUSINESS_UNIT"] == "RSL"] or categories
-        for idx in range(1, count + 1):
-            if idx <= 15 and lng_customers:
-                customer = rng.choice(lng_customers)
-                category = rng.choice([c for c in categories if c["ORDER_TYPE"] == "Project-Based"])
-                amount = rng.uniform(5_000_000, 20_000_000)
-                days_outstanding = rng.randint(61, 89)
-                dispute = "N"
-                reason = None
-            elif idx <= int(count * 0.18) and chart_apac_me:
-                customer = rng.choice(chart_apac_me)
-                category = rng.choice(categories)
-                amount = rng.uniform(70000, 1_200_000)
-                days_outstanding = rng.randint(75, 105)
-                dispute = rng.choice(["N", "N", "Y"])
-                reason = None if dispute == "N" else rng.choice(["Missing PO", "Contract Terms"])
-            elif idx <= int(count * 0.32):
-                customer = rng.choice([c for c in customers if c["INDUSTRY"] == "New Energy"])
-                category = rng.choice(categories)
-                amount = rng.uniform(90000, 2_500_000)
-                days_outstanding = rng.randint(20, 95)
-                dispute = rng.choice(["N", "Y"])
-                reason = None if dispute == "N" else "Contract Terms"
-            elif idx <= int(count * 0.50):
-                customer = rng.choice(customers)
-                category = rng.choice(rsl_categories)
-                amount = rng.uniform(2500, 50000)
-                days_outstanding = rng.randint(61, 140)
-                dispute = rng.choice(["N", "N", "Y"])
-                reason = None if dispute == "N" else rng.choice(["Missing PO", "Quantity Dispute"])
-            else:
-                customer = rng.choice(customers)
-                category = rng.choice(categories)
-                amount = rng.uniform(8000, 950000)
-                days_outstanding = max(0, int(rng.gauss(42 if customer["LEGACY_ORG"] == "Baker Hughes" else 62, 22)))
-                dispute = rng.choices(["N", "Y"], weights=[86, 14], k=1)[0]
-                reason = None if dispute == "N" else rng.choice(["Pricing Discrepancy", "Missing PO", "Quantity Dispute", "Contract Terms"])
-            invoice_date = end - timedelta(days=days_outstanding)
-            term_days = int(customer["CONTRACTED_PAYMENT_TERMS"].split()[1])
-            due_date = invoice_date + timedelta(days=term_days)
-            paid = days_outstanding <= term_days + rng.randint(0, 18)
-            payment_date = invoice_date + timedelta(days=days_outstanding) if paid else None
-            amount_paid = amount if paid else 0.0
-            bucket = self._aging_bucket(max(0, (end - due_date).days if not paid else 0))
-            cash_opportunity = 0.0 if paid else amount * min(max(days_outstanding - 45, 0), 90) / 365.0
-            rows.append({
-                "INVOICE_ID": idx,
-                "INVOICE_DATE": invoice_date.isoformat(),
-                "DUE_DATE": due_date.isoformat(),
-                "CUSTOMER_ID": customer["CUSTOMER_ID"],
-                "CATEGORY_ID": category["CATEGORY_ID"],
-                "INVOICE_AMOUNT_USD": round(amount, 2),
-                "PAYMENT_RECEIVED_DATE": payment_date.isoformat() if payment_date else None,
-                "AMOUNT_PAID_USD": round(amount_paid, 2),
-                "DAYS_OUTSTANDING": days_outstanding,
-                "AGING_BUCKET": bucket,
-                "DISPUTE_FLAG": dispute,
-                "DISPUTE_REASON": reason,
-                "COLLECTOR_ASSIGNED": "Y" if (not paid and days_outstanding > 60 and rng.random() < 0.55) else "N",
-                "LEGACY_ORG": customer["LEGACY_ORG"],
-                "CASH_ACCELERATION_OPPORTUNITY_USD": round(cash_opportunity, 2),
-            })
-        return rows
-
-    @staticmethod
-    def _aging_bucket(days_past_due: int) -> str:
-        if days_past_due <= 0:
-            return "Current"
-        if days_past_due <= 30:
-            return "1-30 Days"
-        if days_past_due <= 60:
-            return "30-60 Days"
-        if days_past_due <= 90:
-            return "60-90 Days"
-        return "90+ Days"

demoprep_app/dataset/generators/router.py

@@ -4,7 +4,6 @@ from __future__ import annotations
 
 from demoprep_app.dataset.contracts import DatasetBundle
 from demoprep_app.dataset.generators.base import ScenarioDatasetGenerator
-from demoprep_app.dataset.generators.post_merger_integration import PostMergerIntegrationDatasetGenerator
 from demoprep_app.dataset.generators.retail_sales import RetailSalesDatasetGenerator
 from demoprep_app.dataset.generators.saas_sales import SaaSSalesDatasetGenerator
 from demoprep_app.dataset.generators.template_generator import TemplateDatasetGenerator
@@ -14,7 +13,6 @@ from demoprep_app.scenario.contract import ScenarioContract
 class DatasetGeneratorRouter:
     def __init__(self, generators: list[ScenarioDatasetGenerator] | None = None) -> None:
         self.generators = generators or [
-            PostMergerIntegrationDatasetGenerator(),
             RetailSalesDatasetGenerator(),
             SaaSSalesDatasetGenerator(),
             TemplateDatasetGenerator(),

demoprep_app/pipeline/dataset_first.py

@@ -8,6 +8,7 @@ from dataclasses import dataclass
 
 from demoprep_app.dataset.company_contract import build_company_data_contract
 from demoprep_app.dataset.contracts import DatasetBundle
+from demoprep_app.dataset.explicit_prompt import build_explicit_prompt_dataset
 from demoprep_app.dataset.generators.router import DatasetGeneratorRouter
 from demoprep_app.ddl import DatasetDdlCompiler
 from demoprep_app.scenario.contract import DimensionSpec, ScenarioContract
@@ -62,11 +63,6 @@ def infer_scenario_type(
 
     if any(tok in text for tok in ("air transport", "airline", "airport", "aircraft", "flight", "passenger", "baggage")) and "finance" in text:
         return "airline_finance"
-    if (
-        any(tok in text for tok in ("post-merger", "post merger", "acquisition", "acquired", "integration", "synergy", "synergies"))
-        and any(tok in text for tok in ("supplier", "spend", "aftermarket", "installed base", "attach rate", "ar aging", "dso", "invoicing"))
-    ):
-        return "post_merger_integration"
     if any(tok in text for tok in ("air transport", "airline", "airport", "aircraft", "flight", "passenger", "baggage")):
         return "airline_route_operations"
     if _has_any_term(text, ("ad yield", "ad monetization", "advertising monetization", "arpu", "ecpm", "fill rate", "ctv", "connected tv", "smart tv", "smartcast", "fast channel", "ad requests", "completion rate")):
@@ -112,6 +108,11 @@ def _has_any_term(text: str, terms: tuple[str, ...]) -> bool:
     return any(re.search(rf"(?<![a-z0-9]){re.escape(term.lower())}(?![a-z0-9])", text) for term in terms)
 
 
+def _is_custom_request(vertical: str | None, function: str | None) -> bool:
+    markers = {value.strip().lower() for value in (vertical or "", function or "") if value.strip()}
+    return "custom" in markers or "* custom *" in markers
+
+
 def build_dataset_first_demo(
     *,
     company_name: str,
@@ -124,7 +125,27 @@ def build_dataset_first_demo(
     llm_model: str | None = None,
     use_llm_contract: bool = False,
 ) -> DatasetFirstBuild | None:
+    seed_text = f"{company_url}|{use_case}".lower()
+    explicit_seed = int(hashlib.sha256(seed_text.encode("utf-8")).hexdigest()[:8], 16)
+    explicit_dataset = build_explicit_prompt_dataset(
+        company_name=company_name,
+        company_url=company_url,
+        use_case=use_case,
+        row_count_guidance=row_count_guidance,
+        seed=explicit_seed,
+    )
+    if explicit_dataset is not None:
+        ddl = DatasetDdlCompiler().compile(explicit_dataset)
+        return DatasetFirstBuild(
+            scenario=explicit_dataset.scenario,
+            dataset=explicit_dataset,
+            ddl=ddl,
+        )
+
+    custom_request = _is_custom_request(vertical, function)
     preliminary_type = infer_scenario_type(use_case, vertical, function, company_name, company_url)
+    if custom_request:
+        preliminary_type = None
     deterministic_types = {
         "saas_sales",
         "retail_sales",
@@ -148,7 +169,6 @@ def build_dataset_first_demo(
         "cpg_finance",
         "portfolio_financials",
         "inventory_supply_chain",
-        "post_merger_integration",
     }
     extraction = extract_dataset_scenario(
         company_name=company_name,
@@ -160,6 +180,19 @@ def build_dataset_first_demo(
         llm_model=llm_model,
         use_llm=use_llm_contract,
     )
+    if custom_request and (extraction is None or extraction.source != "llm"):
+        if use_llm_contract:
+            issue_detail = "; ".join(extraction.constraints[-3:]) if extraction and extraction.constraints else "no LLM contract returned"
+            scenario_name = extraction.scenario_type if extraction else "unknown"
+            confidence = extraction.confidence if extraction else 0.0
+            source = extraction.source if extraction else "none"
+            raise ValueError(
+                "AI dataset contract unavailable for custom demo. "
+                f"Refusing weak fallback contract for scenario '{scenario_name}' "
+                f"(source={source}, confidence={confidence}). "
+                f"{issue_detail}"
+            )
+        return None
     if (
         use_llm_contract
         and preliminary_type not in deterministic_types
@@ -177,6 +210,8 @@ def build_dataset_first_demo(
         scenario_type = preliminary_type
     else:
         scenario_type = extraction.scenario_type if extraction else preliminary_type
+    if scenario_type == "post_merger_integration":
+        return None
     if not scenario_type:
         return None
 

demoprep_app/scenario/extractor.py

@@ -218,8 +218,6 @@ def _fallback_extract(text: str, use_case: str, vertical: str | None, function:
 
 def _infer_from_text(text: str) -> str | None:
     low = text.lower()
-    if _has_any(low, ("post-merger", "post merger", "acquisition", "acquired", "integration", "synergy", "synergies", "installed base", "aftermarket attach", "ar aging", "dso")):
-        return "post_merger_integration"
     if _has_any(low, ("professional services", "consulting", "assurance", "audit", "advisory", "billable", "realized rate", "engagement margin", "staff pyramid", "cross-sell")):
         return "professional_services_engagements"
     if _has_any(low, ("arena", "venue", "stadium", "ticket", "attendance", "fan", "season ticket", "suite", "sponsorship", "basketball", "hockey", "concert", "theater", "theatre")):

demoprep_app/scenario/families.py

@@ -192,20 +192,6 @@ SCENARIO_FAMILIES: dict[str, ScenarioFamilyTemplate] = {
         "operations",
         ("Which lines have the best yield?", "Where is downtime increasing?", "How is throughput trending?"),
     ),
-    "post_merger_integration": ScenarioFamilyTemplate(
-        "post_merger_integration",
-        "FACT_SPEND",
-        "supplier-commodity-site-purchase-order",
-        "SPEND_AMOUNT_USD",
-        ("DIM_SUPPLIER", "DIM_COMMODITY", "DIM_CUSTOMER", "DIM_PRODUCT", "DIM_INVOICE_CATEGORY"),
-        ("Supplier Consolidation", "Aftermarket Attach", "Cross-Sell", "AR Aging", "Cash Flow"),
-        "post_merger_integration",
-        (
-            "Where are duplicate suppliers and fragmented spend creating synergy opportunities?",
-            "Which installed-base cohorts have low aftermarket attach rate?",
-            "Which overdue invoices and regions are trapping working capital?",
-        ),
-    ),
     "automotive_sales": ScenarioFamilyTemplate(
         "automotive_sales",
         "VEHICLE_SALES",

demoprep_app/scenario/selector.py

@@ -16,7 +16,6 @@ def infer_scenario_family(use_case: str, vertical: str | None = None, function:
     text = f"{vertical or ''} {function or ''} {use_case or ''}".lower()
 
     rules: tuple[tuple[str, tuple[str, ...], tuple[str, ...]], ...] = (
-        ("post_merger_integration", ("post-merger", "post merger", "acquisition", "acquired", "integration", "synergy", "synergies", "installed base", "attach rate", "ar aging", "dso"), ("supplier", "spend", "aftermarket", "invoice", "cash flow", "working capital")),
         ("professional_services_engagements", ("professional services", "consulting", "assurance", "audit", "advisory", "billable", "realized rate", "engagement margin", "staff pyramid", "cross-sell"), ("sector", "client", "partner", "utilization", "margin")),
         ("shipping_sales", ("shipping", "parcel", "shipment", "shipments", "delivery service", "express", "ground", "freight"), ("sales", "account", "revenue", "lane", "service")),
         ("trucking_finance", ("trucking", "truckload", "carrier", "driver", "fleet", "load", "loads", "empty mile", "revenue per mile"), ("finance", "fuel", "margin", "cost per mile")),

tests/e2e_quality.py

@@ -2235,7 +2235,24 @@ if __name__ == "__main__":
                         help="Tag for summary filename: 'test' → latest_test_summary.md, 'prod' → latest_prod_summary.md")
     parser.add_argument("--ts-environment", type=str, default="",
                         help="Override the TS Environment dropdown value for this run")
+    parser.add_argument("--test-user", type=str, default="",
+                        help="Override TEST_USER for this run only")
+    parser.add_argument("--test-password", type=str, default="",
+                        help="Override TEST_PASSWORD for this run only; prefer --test-password-env")
+    parser.add_argument("--test-password-env", type=str, default="",
+                        help="Environment variable containing the password for --test-user")
     args = parser.parse_args()
+    if args.test_user:
+        TEST_USER = args.test_user
+        if not args.test_password and not args.test_password_env:
+            raise SystemExit(
+                "--test-user requires --test-password or --test-password-env; "
+                "otherwise the runner would use the default TEST_PASSWORD for a different user."
+            )
+    if args.test_password_env:
+        TEST_PASSWORD = os.getenv(args.test_password_env, "")
+    elif args.test_password:
+        TEST_PASSWORD = args.test_password
     if args.dry_run:
         DRY_RUN = True
     if args.url:

tests/test_dataset_first_builders.py

@@ -118,7 +118,7 @@ def test_custom_dataset_first_refuses_weak_fallback_when_llm_contract_required(m
     assert "Refusing weak fallback contract" in message
 
 
-def test_custom_sports_venue_contract_uses_domain_shape_and_values():
+def test_custom_sports_venue_without_explicit_tables_does_not_use_canned_template():
     use_case = (
         "A live entertainment arena needs analytics across its basketball teams, "
         "Downtown Center, Uptown Theater, concerts, fan demographics, ticket tiers, "
@@ -134,37 +134,10 @@ def test_custom_sports_venue_contract_uses_domain_shape_and_values():
         row_count_guidance=100,
     )
 
-    assert build is not None
-    assert build.scenario.scenario_type == "sports_venue_fan_engagement"
-    assert [table.name for table in build.dataset.tables] == [
-        "DATES",
-        "TEAMS",
-        "VENUES",
-        "EVENTS",
-        "TICKET_TIERS",
-        "FAN_SEGMENTS",
-        "CHANNELS",
-        "EVENT_FAN_ENGAGEMENT",
-    ]
-
-    table_map = build.dataset.table_map()
-    team_names = {row["TEAM_NAME"] for row in table_map["TEAMS"].rows}
-    venue_names = {row["VENUE_NAME"] for row in table_map["VENUES"].rows}
-    channel_names = {row["CHANNEL_NAME"] for row in table_map["CHANNELS"].rows}
-    assert any("Team Alpha" in name for name in team_names)
-    assert any("Downtown Center" in name for name in venue_names)
-    assert any("Email" in name for name in channel_names)
-
-    sample = table_map["EVENT_FAN_ENGAGEMENT"].rows[:50]
-    assert sample
-    for row in sample:
-        assert row["ATTENDANCE"] <= row["CAPACITY"]
-        assert row["PAID_ATTENDANCE"] <= row["ATTENDANCE"]
-        assert row["EMAIL_CLICKS"] <= row["EMAIL_OPENS"] <= row["EMAIL_SENDS"]
-        assert 0 <= row["UTILIZATION_PCT"] <= 100
+    assert build is None
 
 
-def test_revenue_does_not_trigger_venue_scenario_for_professional_services():
+def test_custom_professional_services_without_explicit_tables_does_not_use_canned_template():
     use_case = (
         "Build a professional services demo for EY's assurance and consulting lines. "
         "Track billable hours vs. budget, revenue per sector, staff pyramid health, "
@@ -180,31 +153,7 @@ def test_revenue_does_not_trigger_venue_scenario_for_professional_services():
         row_count_guidance=100,
     )
 
-    assert build is not None
-    assert build.scenario.scenario_type == "professional_services_engagements"
-    assert [table.name for table in build.dataset.tables] == [
-        "MONTHS",
-        "CLIENTS",
-        "SERVICE_LINES",
-        "SECTORS",
-        "CONSULTANTS",
-        "REGIONS",
-        "CLIENT_ENGAGEMENTS",
-    ]
-
-    table_map = build.dataset.table_map()
-    service_lines = {row["SERVICE_LINE_NAME"] for row in table_map["SERVICE_LINES"].rows}
-    assert "Assurance" in service_lines
-    assert "Consulting" in service_lines
-
-    sample = table_map["CLIENT_ENGAGEMENTS"].rows[:50]
-    assert sample
-    for row in sample:
-        assert row["WRITEOFF_USD"] <= row["GROSS_REVENUE_USD"]
-        assert row["REALIZED_REVENUE_USD"] == round(row["GROSS_REVENUE_USD"] - row["WRITEOFF_USD"], 2)
-        assert row["LABOR_COST_USD"] <= row["REALIZED_REVENUE_USD"]
-        assert row["ENGAGEMENT_MARGIN_USD"] == round(row["REALIZED_REVENUE_USD"] - row["LABOR_COST_USD"], 2)
-        assert row["ADVISORY_CROSS_SELLS"] <= row["AUDIT_CLIENTS"]
+    assert build is None
 
 
 def test_quality_pool_domains_do_not_fall_back_to_generic_templates():
@@ -508,7 +457,7 @@ def test_grocery_finance_routes_to_cpg_financials_not_sales_fact():
     assert "LTV_USD" not in fact_columns
 
 
-def test_baker_hughes_chart_post_merger_demo_has_three_value_stories():
+def test_baker_hughes_chart_post_merger_keywords_do_not_trigger_nine_table_template():
     use_case = (
         "Baker Hughes acquired Chart Industries and needs a post-merger integration "
         "demo for supplier spend consolidation, aftermarket attach rate on Chart's "
@@ -523,59 +472,58 @@ def test_baker_hughes_chart_post_merger_demo_has_three_value_stories():
         row_count_guidance=200,
     )
 
+    assert build is None
+
+
+def test_explicit_table_prompt_overrides_post_merger_template_for_ar_aging():
+    use_case = """
+Customer Invoicing & AR Aging
+Generate synthetic accounts receivable and invoicing data for a Baker Hughes + Chart Industries cash flow demo.
+The request is intentionally scoped to the following tables only.
+
+Tables
+
+dim_customer (~200 rows)
+Customer ID, customer name, industry, country, region, customer tier, legacy org, contracted payment terms.
+
+dim_invoice_category (~20 rows)
+Category ID, category name, business unit, order type.
+
+fact_invoices (~25,000 rows, Jan 2025 - Jun 2026)
+Invoice ID, invoice date, due date, customer ID, invoice category ID, invoice amount USD,
+payment received date, amount paid USD, dispute flag, dispute reason, collector assigned, legacy org.
+
+Anomalies to Seed
+APAC and Middle East Chart customers averaging 75-90 days to pay vs BKR benchmark of 45 days.
+Large LNG project invoices sitting 60-90 days overdue.
+"""
+    build = build_dataset_first_demo(
+        company_name="Baker Hughes + Chart Industries",
+        company_url="https://www.bakerhughes.com",
+        use_case=use_case,
+        vertical="* CUSTOM *",
+        function=None,
+        row_count_guidance=1000,
+    )
+
     assert build is not None
-    assert build.scenario.scenario_type == "post_merger_integration"
+    assert build.scenario.scenario_type == "explicit_table_contract"
     assert [table.name for table in build.dataset.tables] == [
-        "DIM_SUPPLIER",
-        "DIM_COMMODITY",
         "DIM_CUSTOMER",
-        "DIM_PRODUCT",
         "DIM_INVOICE_CATEGORY",
-        "FACT_SPEND",
-        "FACT_EQUIPMENT_INSTALLED",
-        "FACT_ORDERS",
         "FACT_INVOICES",
     ]
-    assert build.ddl.count("CREATE TABLE") == 9
-    assert build.ddl.count("FOREIGN KEY") >= 8
+    assert build.ddl.count("CREATE TABLE") == 3
+    assert build.ddl.count("FOREIGN KEY") >= 2
 
     tables = build.dataset.table_map()
-    assert len(tables["DIM_SUPPLIER"].rows) == 300
-    assert len(tables["DIM_COMMODITY"].rows) == 40
     assert len(tables["DIM_CUSTOMER"].rows) == 200
-    assert len(tables["DIM_PRODUCT"].rows) == 150
-    assert len(tables["FACT_SPEND"].rows) >= 1000
-    assert len(tables["FACT_EQUIPMENT_INSTALLED"].rows) >= 100
-    assert len(tables["FACT_ORDERS"].rows) >= 600
-    assert len(tables["FACT_INVOICES"].rows) >= 500
-
-    duplicate_suppliers = [row for row in tables["DIM_SUPPLIER"].rows if row["IS_DUPLICATE_SUPPLIER"]]
-    assert len(duplicate_suppliers) >= 30
-    assert any(row["CONSOLIDATION_OPPORTUNITY_USD"] > 0 for row in tables["FACT_SPEND"].rows)
-
-    chart_gap_installations = [
-        row for row in tables["FACT_EQUIPMENT_INSTALLED"].rows
-        if row["SITE_REGION"] in {"APAC", "Middle East"} and row["SERVICE_CONTRACT_IN_PLACE"] == "N"
-    ]
-    assert chart_gap_installations
-    assert any(row["AFTERMARKET_GAP_USD"] > 0 for row in chart_gap_installations)
-
-    overdue_invoices = [
-        row for row in tables["FACT_INVOICES"].rows
-        if row["AGING_BUCKET"] in {"60-90 Days", "90+ Days"} and row["PAYMENT_RECEIVED_DATE"] is None
-    ]
-    assert overdue_invoices
-    assert any(row["CASH_ACCELERATION_OPPORTUNITY_USD"] > 0 for row in overdue_invoices)
-
-    synthetic_suffix_values = []
-    for table_name in ("DIM_SUPPLIER", "DIM_COMMODITY", "DIM_CUSTOMER", "DIM_PRODUCT"):
-        for row in tables[table_name].rows:
-            for key, value in row.items():
-                if not key.endswith("_NAME"):
-                    continue
-                if isinstance(value, str) and value.split()[-1].isdigit():
-                    synthetic_suffix_values.append(value)
-    assert not synthetic_suffix_values[:5]
+    assert len(tables["DIM_INVOICE_CATEGORY"].rows) == 20
+    assert len(tables["FACT_INVOICES"].rows) == 1000
+    assert {"DAYS_OUTSTANDING", "AGING_BUCKET", "PAST_DUE_AMOUNT_USD"} <= {
+        column.name for column in tables["FACT_INVOICES"].columns
+    }
+    assert any(row["AGING_BUCKET"] in {"60-90 Days", "90+ Days"} for row in tables["FACT_INVOICES"].rows)
 
 
 def test_saas_finance_routes_to_subscription_revenue_not_generic_unit_economics():

thoughtspot_deployer.py

@@ -3479,7 +3479,7 @@ class ThoughtSpotDeployer:
             # Use the enhanced model creation that includes constraint references
             model_tml = self._create_model_with_constraints(tables, foreign_keys, table_guids, table_constraints, model_name, connection_name)
             print(f"\n📄 Model TML being sent:\n{model_tml}")
-            
+
             response = self.session.post(
                 f"{self.base_url}/api/rest/2.0/metadata/tml/import",
                 json={
@@ -3488,6 +3488,62 @@ class ThoughtSpotDeployer:
                     "create_new": True
                 }
             )
+
+            # Some complex multi-fact models are rejected by ThoughtSpot when
+            # the Model TML repeats explicit model-table joins even though the
+            # table objects already have joins from Batch 2. Retry once with the
+            # same tables/columns but no model-table joins.
+            if response.status_code == 200:
+                try:
+                    _model_import_preview = response.json()
+                    _preview_objects = self._normalize_tml_import_response_objects(_model_import_preview) or []
+                    _first_status = (
+                        _preview_objects[0].get('response', {}).get('status', {})
+                        if _preview_objects else {}
+                    )
+                    _first_status_code = str(_first_status.get('status_code') or '')
+                    _first_error = str(_first_status.get('error_message') or '')
+                    _first_error_code = str(_first_status.get('error_code') or '')
+                except Exception:
+                    _first_status_code = ''
+                    _first_error = ''
+                    _first_error_code = ''
+
+                if (
+                    _first_status_code == 'ERROR'
+                    and (
+                        _first_error_code == '13122'
+                        or 'schema validation failed' in _first_error.lower()
+                    )
+                ):
+                    warning = (
+                        "Model import with explicit joins failed schema validation; "
+                        "retrying model import without model-table joins."
+                    )
+                    log_progress(f"   ⚠️  {warning}")
+                    results['warnings'].append(warning)
+                    if _slog:
+                        _slog.log(
+                            "thoughtspot",
+                            "model import retrying without joins",
+                            error_code=_first_error_code,
+                            error=_first_error[:500],
+                        )
+                    model_tml = self.create_actual_model_tml(
+                        tables,
+                        foreign_keys,
+                        table_guids=table_guids,
+                        model_name=model_name,
+                        connection_name=connection_name,
+                    )
+                    response = self.session.post(
+                        f"{self.base_url}/api/rest/2.0/metadata/tml/import",
+                        json={
+                            "metadata_tmls": [model_tml],
+                            "import_policy": "ALL_OR_NONE",
+                            "create_new": True
+                        }
+                    )
             
             if response.status_code == 200:
                 result = response.json()