Changes for the intent

README.md

@@ -96,24 +96,26 @@ reset(dataset, task_id, intent)
 
 ## Scoring Formulas
 
+**Easy scores most** — the basic cleaning foundation counts for the most weight in every cascade.
+
 ### Easy
 ```
-score = resolution(null + type_error + whitespace + category) × 0.70
-      + validation × 0.30
+score = resolution(null + type_error + whitespace + category) × 0.80
+      + validation × 0.20
       − fp_penalty
 ```
 
-### Medium
+### Medium  (easy = 70% of the score)
 ```
-score = easy_score × 0.40
-      + avg(constraint_resolved + outlier_resolved) × 0.60
+score = easy_score × 0.70
+      + avg(constraint_resolved + outlier_resolved + duplicate_resolved) × 0.30
       − fp_penalty
 ```
 
-### Hard
+### Hard  (medium = 70% of the score)
 ```
 # With FK violations:
-score = medium_score × 0.50 + fk_resolved × 0.50 + reasoning_bonus − trap_penalty
+score = medium_score × 0.70 + fk_resolved × 0.30 + reasoning_bonus − trap_penalty
 
 # Single-table dataset (no FK violations to find):
 score = medium_score + reasoning_bonus − trap_penalty
@@ -220,6 +222,36 @@ Dense per-step rewards — every action returns a signal:
 
 ---
 
+## Multiple Tables
+
+Upload **2–3 files** in the Agent Demo tab. They are merged into a multi-sheet XLSX — each sheet becomes a separate SQLite table. The agent will:
+
+1. `select_tables` — list all available tables
+2. Inspect schemas and attempt `join_tables` with the best matching key column
+3. Profile and bulk-fix each table independently
+
+**Supported formats per file:** CSV, JSON, JSONL, Parquet, XLSX
+
+Alternatively, upload a **single XLSX with multiple sheets** — each sheet loads as a separate table automatically.
+
+---
+
+## Business Query Feature
+
+For the **Business Analytics** intent, enter a SQL `SELECT` query in the UI. After the agent finishes cleaning (Hard task), the query runs on the cleaned in-memory database and results are displayed in the UI.
+
+```sql
+-- Example
+SELECT "Pclass", COUNT(*) AS passengers, AVG("Fare") AS avg_fare
+FROM dataset
+GROUP BY "Pclass"
+ORDER BY avg_fare DESC
+```
+
+Works on any intent — not just business_query.
+
+---
+
 ## Quick Start
 
 ### Docker (recommended)

inference.py

@@ -85,10 +85,14 @@ def log_start(task: str, model: str) -> None:
 def log_step(step: int, action: str, reward: float, done: bool,
              error: Optional[str] = None) -> None:
     action_str = action.replace("\n", " ").replace("\r", " ").strip()[:120]
+    error_str  = (
+        error.replace("\n", " ").replace("\r", " ").strip()[:120]
+        if error else "null"
+    )
     print(
         f"[STEP] step={step} action={action_str} "
         f"reward={reward:.2f} done={str(done).lower()} "
-        f"error={error if error else 'null'}",
+        f"error={error_str}",
         flush=True,
     )
 
@@ -195,10 +199,10 @@ def run_task(task_id: str, intent: str = "") -> float:
 
     budget      = STEP_BUDGETS[task_id]
     rewards: list[float] = []
-    steps_taken = 0
+    step_num    = 0        # initialised here so except/finally can always read it
     score       = 0.0
     success     = False
-    env         = None   # initialised inside try so finally can check
+    env         = None     # initialised inside try so finally can check
 
     log_start(task=task_id, model=MODEL_NAME)
 
@@ -216,8 +220,7 @@ def run_task(task_id: str, intent: str = "") -> float:
         columns      = obs.tables_summary.get(table, {}).get("columns", [])
         visible_cols = [c for c in columns if c not in ("id", "_source_format")]
 
-        done     = False
-        step_num = 0
+        done         = False
         col_profiles: dict[str, dict] = {}
 
         llm_messages = [{"role": "system", "content": SYSTEM_PROMPT}]
@@ -322,13 +325,12 @@ def run_task(task_id: str, intent: str = "") -> float:
             positive = sum(r for r in rewards if r > 0)
             score = max(0.0, min(1.0, positive / max(budget * 0.15, 0.01)))
 
-        success     = score >= 0.50
-        steps_taken = step_num
+        success = score >= 0.50
 
     except Exception as exc:
         # Catch-all — log a terminal error step so judges see what failed
         log_step(
-            max(steps_taken + 1, 1), "error", 0.0, True,
+            max(step_num + 1, 1), "error", 0.0, True,
             f"{type(exc).__name__}: {str(exc)[:60]}",
         )
 
@@ -339,7 +341,7 @@ def run_task(task_id: str, intent: str = "") -> float:
                 env.close()
             except Exception:
                 pass
-        log_end(success=success, steps=steps_taken, rewards=rewards)
+        log_end(success=success, steps=step_num, rewards=rewards)
 
     return score
 

sqlsherlock_env/server/graders/task2.py

@@ -88,5 +88,6 @@ def grade(
     )
 
     combined = (c_score + o_score + d_score) / 3.0
-    raw = t1 * 0.40 + combined * 0.60 - fp_penalty
+    # Easy (task1) gets the highest weight — basic cleaning matters most.
+    raw = t1 * 0.70 + combined * 0.30 - fp_penalty
     return max(0.0, min(1.0, round(raw, 4)))

sqlsherlock_env/server/graders/task3.py

@@ -82,18 +82,18 @@ def grade(
     # again here to avoid double-counting.
 
     if fk_issues:
-        # Dataset has FK violations — score FK resolution as 50% of hard grade
+        # Medium (task2) gets highest weight — medium cleaning is the foundation.
+        # FK resolution adds the hard-level component at 30%.
         fk_score, _ = _resolution_score(fk_issues, cleaned_rows, removed_ids, pk_col, db)
         raw = (
-            t2        * 0.50
-            + fk_score  * 0.50
+            t2        * 0.70
+            + fk_score  * 0.30
             + r_bonus
             - trap_pen
         )
     else:
         # Single-table dataset — no FK issues to resolve.
-        # Hard difficulty is demonstrated entirely through trap avoidance and reasoning.
-        # Grade purely on medium performance + trap + reasoning (no free FK credit).
+        # Hard difficulty demonstrated through trap avoidance and reasoning.
         raw = t2 + r_bonus - trap_pen
 
     return max(0.0, min(1.0, round(raw, 4)))

sqlsherlock_env/server/requirements.txt

@@ -9,3 +9,4 @@ pandas>=2.0.0
 pyarrow>=14.0.0
 gradio>=4.44.0
 scikit-learn>=1.4.0
+openpyxl>=3.1.0

sqlsherlock_env/server/ui.py

@@ -2,28 +2,28 @@
 SQLSherlock-Env dashboard UI.
 
 User flow:
-  1. Provide a dataset (file upload or HuggingFace name).
-  2. Select ONE cleaning intent (visualization / ML training / business analytics).
-  3. Select output format (CSV / JSON / Parquet).
-  4. The agent runs 3 escalating difficulty tasks for that intent:
-       Easy → Medium → Hard
-  5. Live step log streams as each task runs.
-  6. Download the cleaned file from the Hard task when complete.
+  1. Provide dataset(s) — upload up to 3 files OR enter a HuggingFace name.
+  2. Select ONE cleaning intent.
+  3. Describe your specific requirement (dashboard goal / model goal / business question).
+  4. (Optional) enter a SQL query for direct business analytics.
+  5. The agent runs 3 escalating tasks: Easy → Medium → Hard.
+  6. Live step log, before/after comparison, and a final answer are shown.
+  7. Download the cleaned file from the Hard task.
 """
 
-import os
 import re
 import tempfile
-from typing import Generator
+from pathlib import Path
+from typing import Any, Generator
 
 import gradio as gr
+import pandas as pd
 
 
 # ---------------------------------------------------------------------------
 # Constants
 # ---------------------------------------------------------------------------
 
-# Each intent maps to its 3 difficulty task IDs (easy → medium → hard)
 INTENT_TASK_MAP: dict[str, list[str]] = {
     "visualization":  ["viz_easy",  "viz_medium",  "viz_hard"],
     "ml_training":    ["ml_easy",   "ml_medium",   "ml_hard"],
@@ -36,12 +36,6 @@ INTENT_LABELS: dict[str, str] = {
     "business_query": "Business Analytics — clean for SQL reporting & queries",
 }
 
-DIFFICULTY_LABELS: dict[str, str] = {
-    "easy":   "Easy   (30 steps) — nulls, type errors, whitespace, categories",
-    "medium": "Medium (40 steps) — + constraints, outliers",
-    "hard":   "Hard   (50 steps) — + duplicates, FK violations, trap",
-}
-
 INTENT_CHOICES = [
     ("Visualization — dashboards & charts",          "visualization"),
     ("ML Training — model training pipelines",       "ml_training"),
@@ -50,33 +44,67 @@ INTENT_CHOICES = [
 
 FORMAT_CHOICES = ["csv", "json", "parquet"]
 
-_EXAMPLE_HF_DATASETS = [
-    ["phihung/titanic"],
-    ["scikit-learn/iris"],
-]
+_EXAMPLE_HF = [["phihung/titanic"], ["scikit-learn/iris"]]
 
 _AGENT_EXAMPLES = [
-    ["phihung/titanic", "visualization",  "csv"],
-    ["phihung/titanic", "ml_training",    "csv"],
-    ["phihung/titanic", "business_query", "csv"],
+    ["phihung/titanic", "visualization",
+     "Build a dashboard showing survival rates by passenger class, gender and age group.", "csv", ""],
+    ["phihung/titanic", "ml_training",
+     "Train a survival prediction model using Age, Fare and Pclass as features.", "csv", ""],
+    ["phihung/titanic", "business_query",
+     "What is the average fare and survival rate for each passenger class?", "csv",
+     'SELECT "Pclass", COUNT(*) AS passengers, ROUND(AVG("Fare"),2) AS avg_fare, ROUND(AVG("Survived"),2) AS survival_rate FROM dataset GROUP BY "Pclass" ORDER BY "Pclass"'],
 ]
 
+# Requirement placeholders per intent
+_REQ_CONFIG: dict[str, dict] = {
+    "visualization": {
+        "label": "Describe your dashboard (what do you want to visualise?)",
+        "placeholder": (
+            "e.g. I want to build a dashboard showing passenger survival rates "
+            "by class, gender and age group. Key metrics: survival %, average fare, "
+            "passenger count per category."
+        ),
+    },
+    "ml_training": {
+        "label": "Describe your model (goal, features, target column)",
+        "placeholder": (
+            "e.g. Building a binary classification model to predict passenger survival. "
+            "Features: Age, Fare, Pclass, Sex. Target: Survived. "
+            "Need clean numeric features with no nulls or extreme outliers."
+        ),
+    },
+    "business_query": {
+        "label": "What business question do you want answered?",
+        "placeholder": (
+            "e.g. What is the average ticket fare for each passenger class? "
+            "Which class had the highest survival rate? "
+            "How many passengers embarked from each port?"
+        ),
+    },
+}
+
+_SQL_PLACEHOLDER = (
+    'SELECT "Pclass", COUNT(*) AS passengers,\n'
+    '       ROUND(AVG("Fare"), 2) AS avg_fare,\n'
+    '       ROUND(AVG("Survived"), 2) AS survival_rate\n'
+    'FROM dataset\n'
+    'GROUP BY "Pclass"\n'
+    'ORDER BY "Pclass"'
+)
+
 
 # ---------------------------------------------------------------------------
 # Helpers
 # ---------------------------------------------------------------------------
 
-def _resolve_source(file_path: str | None, hf_name: str) -> tuple[str | None, str | None]:
-    """Return (source, error_message). File upload takes precedence over HF name."""
-    if file_path:
-        return file_path, None
+def _resolve_source(file1: str | None, hf_name: str) -> tuple[str | None, str | None]:
+    if file1:
+        return file1, None
     name = (hf_name or "").strip()
     if name:
         return name, None
-    return None, (
-        "**No dataset provided.**  "
-        "Upload a file (Option A) or enter a HuggingFace dataset name (Option B)."
-    )
+    return None, "**No dataset provided.** Upload a file or enter a HuggingFace dataset name."
 
 
 def _friendly_error(exc: Exception, source: str) -> str:
@@ -87,42 +115,310 @@ def _friendly_error(exc: Exception, source: str) -> str:
         return "Could not parse the file. Supported: CSV, JSON, JSONL, Parquet, XLSX."
     if "connection" in msg.lower() or "refused" in msg.lower():
         return "Cannot reach the server. Make sure the environment is running."
-    return f"Failed to load dataset ({type(exc).__name__}). Check the name and try again."
+    return f"Error ({type(exc).__name__}): {str(exc)[:150]}"
+
+
+def _read_file(path: str) -> pd.DataFrame:
+    ext = Path(path).suffix.lower()
+    if ext == ".csv":
+        return pd.read_csv(path)
+    if ext in (".json", ".jsonl"):
+        try:
+            return pd.read_json(path)
+        except Exception:
+            return pd.read_json(path, lines=True)
+    if ext == ".parquet":
+        return pd.read_parquet(path)
+    if ext == ".xlsx":
+        return pd.read_excel(path, sheet_name=0)
+    return pd.read_csv(path)
+
+
+def _merge_to_xlsx(files: list[str]) -> str:
+    """Merge 2–3 uploaded files into one multi-sheet XLSX.
+    Each file → one sheet; sheet name = filename stem (≤31 chars).
+    Returns path to the temp file.
+    """
+    tmp = tempfile.NamedTemporaryFile(suffix=".xlsx", delete=False)
+    tmp.close()
+    with pd.ExcelWriter(tmp.name, engine="openpyxl") as writer:
+        for i, fpath in enumerate(files):
+            df = _read_file(fpath)
+            sheet = (Path(fpath).stem[:31] or f"table{i+1}")
+            df.to_excel(writer, sheet_name=sheet, index=False)
+    return tmp.name
+
+
+def _rows_to_df(rows: list[dict]) -> pd.DataFrame:
+    if not rows:
+        return pd.DataFrame()
+    exclude = {"_source_format"}
+    cols = [c for c in rows[0].keys() if c not in exclude]
+    return pd.DataFrame([{c: r.get(c, "") for c in cols} for r in rows])
+
+
+def _df_to_md(df: pd.DataFrame, max_rows: int = 25) -> str:
+    """Convert DataFrame to a markdown table (no external dependencies)."""
+    if df.empty:
+        return "_No results_"
+    display = df.head(max_rows)
+    cols = list(display.columns)
+    header = "| " + " | ".join(str(c) for c in cols) + " |"
+    sep    = "| " + " | ".join("---" for _ in cols) + " |"
+    body   = [
+        "| " + " | ".join(str(display.iloc[i][c]) for c in cols) + " |"
+        for i in range(len(display))
+    ]
+    tail = f"\n_{len(df) - max_rows} more rows…_" if len(df) > max_rows else ""
+    return "\n".join([header, sep] + body) + tail
+
+
+def _schema_md(tables_summary: dict) -> str:
+    if not tables_summary or len(tables_summary) <= 1:
+        return ""
+    lines = ["### Loaded Tables\n"]
+    for tname, info in tables_summary.items():
+        cols   = [c for c in info.get("columns", []) if c != "_source_format"]
+        dtypes = info.get("dtypes", {})
+        n_rows = info.get("row_count", "?")
+        lines.append(f"**`{tname}`** — {n_rows} rows, {len(cols)} columns\n")
+        lines.append("| Column | Type |")
+        lines.append("|--------|------|")
+        for c in cols:
+            lines.append(f"| `{c}` | {dtypes.get(c, '?')} |")
+        lines.append("")
+    return "\n".join(lines)
+
+
+def _nl_to_sql(question: str, table: str, columns: list[str]) -> str | None:
+    """Convert a natural-language business question to a best-effort SQL query.
+
+    Returns None if the question is already SQL or no pattern is detected.
+    """
+    q = question.strip()
+    if not q:
+        return None
+    if q.upper().startswith("SELECT"):
+        return q  # already SQL
+
+    q_low = q.lower()
+
+    # Identify likely numeric and categorical columns
+    num_cols = [c for c in columns if c not in ("id", "_source_format")
+                and any(w in c.lower() for w in
+                        ["fare", "age", "price", "amount", "rate", "score", "salary",
+                         "revenue", "cost", "value", "petal", "sepal", "weight",
+                         "height", "income", "fee", "count", "survived"])]
+    cat_cols = [c for c in columns if c not in ("id", "_source_format")
+                and any(w in c.lower() for w in
+                        ["class", "sex", "gender", "type", "category", "status",
+                         "species", "department", "region", "country", "city",
+                         "embarked", "cabin", "name", "group", "tier"])]
+
+    agg_words  = ["average", "avg", "mean", "total", "sum", "rate",
+                  "how many", "count", "number of", "maximum", "minimum",
+                  "highest", "lowest", "most", "least"]
+    grp_words  = [" by ", " per ", " each ", " for each ", " across ", " group "]
+    has_agg    = any(w in q_low for w in agg_words)
+    has_group  = any(w in q_low for w in grp_words)
+
+    # Detect which column names are mentioned in the question
+    mentioned_num = [c for c in num_cols if c.lower() in q_low]
+    mentioned_cat = [c for c in cat_cols if c.lower() in q_low]
+
+    # Pattern: "average X by Y" / "count per Y"
+    if has_group:
+        grp_col = mentioned_cat[0] if mentioned_cat else (cat_cols[0] if cat_cols else None)
+        agg_col = mentioned_num[0] if mentioned_num else (num_cols[0] if num_cols else None)
+        if grp_col:
+            if agg_col and has_agg:
+                # Multiple aggregations for richness
+                return (
+                    f'SELECT "{grp_col}", COUNT(*) AS count, '
+                    f'ROUND(AVG("{agg_col}"), 2) AS avg_{agg_col.lower()}, '
+                    f'ROUND(MIN("{agg_col}"), 2) AS min_{agg_col.lower()}, '
+                    f'ROUND(MAX("{agg_col}"), 2) AS max_{agg_col.lower()} '
+                    f'FROM "{table}" GROUP BY "{grp_col}" ORDER BY count DESC LIMIT 20'
+                )
+            else:
+                return (
+                    f'SELECT "{grp_col}", COUNT(*) AS count '
+                    f'FROM "{table}" GROUP BY "{grp_col}" ORDER BY count DESC LIMIT 20'
+                )
+
+    # Pattern: "top N …" or "highest / most"
+    top_m = re.search(r'\btop\s+(\d+)\b', q_low)
+    if top_m:
+        n = top_m.group(1)
+        sort_col = mentioned_num[0] if mentioned_num else (num_cols[0] if num_cols else "id")
+        return f'SELECT * FROM "{table}" ORDER BY "{sort_col}" DESC LIMIT {n}'
+
+    if any(w in q_low for w in ["highest", "most", "maximum", "largest"]):
+        sort_col = mentioned_num[0] if mentioned_num else (num_cols[0] if num_cols else "id")
+        return f'SELECT * FROM "{table}" ORDER BY "{sort_col}" DESC LIMIT 10'
+
+    if any(w in q_low for w in ["lowest", "least", "minimum", "smallest"]):
+        sort_col = mentioned_num[0] if mentioned_num else (num_cols[0] if num_cols else "id")
+        return f'SELECT * FROM "{table}" ORDER BY "{sort_col}" ASC LIMIT 10'
+
+    # Fallback: overall summary of numeric columns
+    if cat_cols:
+        c = cat_cols[0]
+        return (
+            f'SELECT "{c}", COUNT(*) AS count '
+            f'FROM "{table}" GROUP BY "{c}" ORDER BY count DESC LIMIT 20'
+        )
+
+    return None
+
+
+def _generate_final_answer(
+    intent: str,
+    user_requirement: str,
+    issues_total: int,
+    issues_remaining: int,
+    columns_fixed: list[str],
+    query_df: pd.DataFrame,
+    auto_sql: str | None,
+    table: str,
+    columns: list[str],
+) -> str:
+    """Build the Final Answer markdown block shown at the end of all tasks."""
+
+    resolved = max(0, issues_total - issues_remaining)
+    fixed_cols_unique = sorted(set(columns_fixed))
+
+    # --- Cleaning summary ---
+    if issues_total == 0:
+        clean_summary = "Dataset had no detectable issues — already clean."
+    elif resolved == issues_total:
+        clean_summary = (
+            f"**All {issues_total} data quality issues resolved** "
+            + (f"across columns: `{'`, `'.join(fixed_cols_unique)}`." if fixed_cols_unique else ".")
+        )
+    else:
+        pct = round(100 * resolved / max(issues_total, 1))
+        clean_summary = (
+            f"**{resolved}/{issues_total} issues resolved ({pct}%)** "
+            + (f"across columns: `{'`, `'.join(fixed_cols_unique)}`." if fixed_cols_unique else ".")
+            + (f" {issues_remaining} issue(s) remain." if issues_remaining > 0 else "")
+        )
+
+    md = ["## Final Answer\n"]
+    md.append(f"### Cleaning Summary\n{clean_summary}\n")
+
+    safe_cols = [c for c in columns if c not in ("id", "_source_format")]
+
+    # --- Intent-specific section ---
+    if intent == "visualization":
+        req = user_requirement.strip() if user_requirement else "No description provided."
+        md.append(f"### Dashboard Requirement\n_{req}_\n")
+        num_cols = [c for c in safe_cols
+                    if any(w in c.lower() for w in
+                           ["fare", "age", "price", "rate", "score", "count", "survived",
+                            "amount", "value", "petal", "sepal"])]
+        cat_cols = [c for c in safe_cols if c not in num_cols]
+        if num_cols:
+            md.append(f"**Recommended metric columns:** `{'`, `'.join(num_cols[:6])}`")
+        if cat_cols:
+            md.append(f"\n**Recommended dimension columns:** `{'`, `'.join(cat_cols[:6])}`")
+        md.append(
+            "\n\nThe dataset is now ready for your dashboard — "
+            "nulls filled, types corrected, categories normalised."
+        )
+
+    elif intent == "ml_training":
+        req = user_requirement.strip() if user_requirement else "No description provided."
+        md.append(f"### Model Requirement\n_{req}_\n")
+        feature_candidates = [c for c in safe_cols
+                               if c.lower() not in ("name", "cabin", "ticket", "survived")]
+        target_candidates  = [c for c in safe_cols if "survived" in c.lower() or "target" in c.lower()]
+        if feature_candidates:
+            md.append(f"**Suggested feature columns:** `{'`, `'.join(feature_candidates[:8])}`")
+        if target_candidates:
+            md.append(f"\n**Likely target column:** `{target_candidates[0]}`")
+        md.append(
+            "\n\nThe dataset is now ready for ML training — "
+            "nulls imputed with column medians/modes, outliers corrected, "
+            "negative constraint violations fixed."
+        )
+
+    elif intent == "business_query":
+        req = user_requirement.strip() if user_requirement else "No question provided."
+        md.append(f"### Business Question\n_{req}_\n")
+
+        if not query_df.empty:
+            md.append(f"### Answer\n")
+            md.append(_df_to_md(query_df))
+            md.append("\n")
+            # Generate a brief narrative from the first row
+            try:
+                cols_list = list(query_df.columns)
+                if len(cols_list) >= 2 and len(query_df) > 0:
+                    top = query_df.iloc[0]
+                    insights = []
+                    for c in cols_list[1:4]:
+                        val = top[c]
+                        if isinstance(val, float):
+                            val = round(val, 2)
+                        insights.append(f"{c} = **{val}**")
+                    md.append(
+                        f"\n**Top result** — {cols_list[0]} = **{top[cols_list[0]]}**: "
+                        + ", ".join(insights) + "\n"
+                    )
+            except Exception:
+                pass
+
+            if auto_sql:
+                md.append(f"\n<details><summary>SQL used</summary>\n\n```sql\n{auto_sql}\n```\n</details>\n")
+        else:
+            md.append(
+                "\n_No query results available. "
+                "Add a SQL query in the **Business SQL Query** field and run again, "
+                "or make sure your question contains group-by keywords like "
+                "\"by class\", \"per gender\", \"average fare\"._\n"
+            )
+
+    md.append(
+        f"\n---\n*Cleaned data available via the Download button above. "
+        f"Table: `{table}` · Columns: {len(safe_cols)}*"
+    )
+    return "\n".join(md)
 
 
 # ---------------------------------------------------------------------------
 # Tab 1: Data Quality Scanner
 # ---------------------------------------------------------------------------
 
-def scan_dataset(file_path: str | None, hf_name: str) -> tuple[str, list, list, list, list]:
+def scan_dataset(file_path: str | None, hf_name: str):
     from server.database import DatabaseEngine
 
     source, err = _resolve_source(file_path, hf_name)
     if err:
         return err, [], [], [], []
-
     try:
-        db = DatabaseEngine(task_id="bq_hard", seed=42, dataset_source=source, max_rows=500)
+        db = DatabaseEngine(task_id="bq_hard", seed=42,
+                            dataset_source=source, max_rows=500)
     except Exception as exc:
         return f"**Error:** {_friendly_error(exc, source)}", [], [], [], []
 
-    table        = db.primary_table
-    issues       = db.issue_registry
-    profiles     = db._profiles.get(table, {})
-    total_rows   = len(db.rows(table))
-    total_cols   = sum(1 for c in profiles if c not in ("id", "_source_format"))
-    total_issues = len(issues)
-    cells        = max(total_rows * total_cols, 1)
-    quality_pct  = max(0.0, min(1.0, 1.0 - total_issues / cells)) * 100
-    grade        = ("Excellent" if quality_pct >= 90 else
-                    "Good"      if quality_pct >= 70 else
-                    "Fair"      if quality_pct >= 50 else "Poor")
+    table       = db.primary_table
+    issues      = db.issue_registry
+    profiles    = db._profiles.get(table, {})
+    total_rows  = len(db.rows(table))
+    total_cols  = sum(1 for c in profiles if c not in ("id", "_source_format"))
+    n_issues    = len(issues)
+    cells       = max(total_rows * total_cols, 1)
+    quality_pct = max(0.0, min(1.0, 1.0 - n_issues / cells)) * 100
+    grade       = ("Excellent" if quality_pct >= 90 else
+                   "Good"      if quality_pct >= 70 else
+                   "Fair"      if quality_pct >= 50 else "Poor")
 
     type_counts: dict[str, int] = {}
     for iss in issues:
         type_counts[iss.issue_type] = type_counts.get(iss.issue_type, 0) + 1
 
-    breakdown_lines = "\n".join(
+    breakdown = "\n".join(
         f"| `{t}` | {c} |"
         for t, c in sorted(type_counts.items(), key=lambda x: -x[1])
     )
@@ -134,14 +430,14 @@ def scan_dataset(file_path: str | None, hf_name: str) -> tuple[str, list, list,
 | **Table** | `{table}` |
 | **Rows** | {total_rows:,} |
 | **Columns** | {total_cols} |
-| **Issues found** | {total_issues} |
+| **Issues found** | {n_issues} |
 | **Quality score** | **{quality_pct:.1f}% — {grade}** |
 
 ### Issue Breakdown
 
 | Issue Type | Count |
 |------------|-------|
-{breakdown_lines}
+{breakdown}
 """
 
     col_rows = []
@@ -152,10 +448,11 @@ def scan_dataset(file_path: str | None, hf_name: str) -> tuple[str, list, list,
         col_issues = sum(1 for i in issues if i.column == col)
         health = ("Good" if null_rate < 0.05 and col_issues == 0 else
                   "Fair" if null_rate < 0.30 and col_issues <= 2 else "Poor")
-        col_rows.append([col, p.get("dtype", "?"), f"{null_rate*100:.1f}%", col_issues, health])
+        col_rows.append([col, p.get("dtype", "?"),
+                         f"{null_rate*100:.1f}%", col_issues, health])
 
-    issue_rows   = [[t, c] for t, c in sorted(type_counts.items(), key=lambda x: -x[1])]
-    sample_rows  = [
+    issue_rows  = [[t, c] for t, c in sorted(type_counts.items(), key=lambda x: -x[1])]
+    sample_rows = [
         [iss.issue_type, iss.column or "-", str(iss.row_id),
          f"{iss.confidence:.0%}",
          "delete row" if iss.correct is None else str(iss.correct)[:40]]
@@ -169,70 +466,105 @@ def scan_dataset(file_path: str | None, hf_name: str) -> tuple[str, list, list,
 
 
 # ---------------------------------------------------------------------------
-# Tab 2: Agent Demo — one intent → 3 difficulty tasks streamed sequentially
+# Tab 2: Agent Demo
 # ---------------------------------------------------------------------------
 
 def run_agent_streaming(
-    file_path: str | None,
+    file1: str | None,
+    file2: str | None,
+    file3: str | None,
     hf_name: str,
     intent: str,
     output_format: str,
-) -> Generator[tuple[str, str, str | None], None, None]:
-    """
-    Yields (log_text, summary_md, download_path) after every action.
-    Runs 3 escalating tasks (easy → medium → hard) for the selected intent.
-    """
+    user_requirement: str,
+    biz_sql: str,
+) -> Generator[tuple[str, str, str | None, Any, Any, str, Any, str], None, None]:
+    """Yields (log, summary_md, download_path, before_df, after_df,
+               schema_md, query_df, final_answer_md)."""
     from server.environment import SQLSherlockEnvironment
     from models import SQLSherlockAction
 
-    source, err = _resolve_source(file_path, hf_name)
-    if err:
-        yield err, "", None
-        return
+    # --- Resolve source ---
+    uploaded = [f for f in [file1, file2, file3] if f]
+    if len(uploaded) >= 2:
+        try:
+            source = _merge_to_xlsx(uploaded)
+            multi  = True
+        except Exception as exc:
+            _empty = pd.DataFrame()
+            yield (f"Could not merge files: {exc}", "", None,
+                   _empty, _empty, "", _empty, "")
+            return
+    elif len(uploaded) == 1:
+        source, multi = uploaded[0], False
+    else:
+        source, err = _resolve_source(None, hf_name)
+        multi = False
+        if not source:
+            _empty = pd.DataFrame()
+            yield (err or "No dataset provided.", "", None,
+                   _empty, _empty, "", _empty, "")
+            return
 
     task_ids     = INTENT_TASK_MAP.get(intent, INTENT_TASK_MAP["visualization"])
     intent_label = INTENT_LABELS.get(intent, intent)
 
-    log_lines: list[str] = []
-    all_scores: list[tuple[str, float]] = []   # [(task_id, score)]
+    log_lines:  list[str] = []
+    all_scores: list[tuple[str, float]] = []
     download_path: str | None = None
-    summary_md = "_Running…_"
+    summary_md     = "_Running…_"
+    before_df      = pd.DataFrame()
+    after_df       = pd.DataFrame()
+    schema_md      = ""
+    query_df       = pd.DataFrame()
+    final_answer   = ""
+    auto_sql_used: str | None = None
+
+    # Track cleaning stats across all tasks (Hard task is what matters for final answer)
+    hard_issues_total    = 0
+    hard_issues_remain   = 0
+    hard_columns_fixed:  list[str] = []
+    hard_table           = ""
+    hard_columns:        list[str] = []
 
     def _log(*parts: str) -> None:
-        log_lines.append(" ".join(parts))
+        log_lines.append(" ".join(str(p) for p in parts))
 
-    def _current_log() -> str:
-        return "\n".join(log_lines)
+    def emit():
+        return ("\n".join(log_lines), summary_md, download_path,
+                before_df, after_df, schema_md, query_df, final_answer)
 
+    src_label = f"{source} ({len(uploaded)} tables merged)" if multi else source
     _log(f"Intent : {intent_label}")
+    _log(f"Dataset: {src_label}")
     _log(f"Format : {output_format}")
-    _log(f"Dataset: {source}")
+    if user_requirement.strip():
+        _log(f"Goal   : {user_requirement.strip()[:100]}")
     _log("")
-    yield _current_log(), summary_md, None
+    yield emit()
 
     for task_id in task_ids:
         difficulty = task_id.split("_")[-1].upper()
         _log(f"{'─'*60}")
         _log(f"[{difficulty}] {task_id}")
         _log(f"{'─'*60}")
-        yield _current_log(), summary_md, None
+        yield emit()
 
-        env   = SQLSherlockEnvironment()
-        score = 0.0
-        step  = 0
-        done  = False
-        obs   = None
+        env     = SQLSherlockEnvironment()
+        score   = 0.0
+        step    = 0
+        done    = False
+        obs     = None
         rewards: list[float] = []
-        fixed_cols: list[str] = []
+        cols_fixed_this_task: list[str] = []
 
         def _do(action_dict: dict) -> float:
             nonlocal step, done, obs
             if done:
                 return 0.0
             step += 1
-            action = SQLSherlockAction(
-                **{k: v for k, v in action_dict.items() if v is not None}
-            )
+            filtered = {k: v for k, v in action_dict.items() if v is not None}
+            action = SQLSherlockAction(**filtered)
             reward = 0.0
             try:
                 obs    = env.step(action)
@@ -250,18 +582,51 @@ def run_agent_streaming(
                 intent=intent,
                 output_format=output_format,
             )
-            table   = list(obs.tables_summary.keys())[0]
-            summary = obs.tables_summary[table]
-            visible = [c for c in summary.get("columns", [])
-                       if c not in ("id", "_source_format")]
-
-            _log(f"  Table: {table}  ({summary.get('row_count','?')} rows, {len(visible)} cols)")
-            yield _current_log(), summary_md, None
+            all_tables = list(obs.tables_summary.keys())
+            table      = all_tables[0]
+            t_info     = obs.tables_summary[table]
+            visible    = [c for c in t_info.get("columns", [])
+                          if c not in ("id", "_source_format")]
+
+            # Schema for multi-table datasets (built once)
+            if len(all_tables) > 1 and not schema_md:
+                schema_md = _schema_md(obs.tables_summary)
+
+            # Before state captured from Easy task (first run only)
+            if task_id == task_ids[0] and hasattr(env, "_db") and env._db:
+                before_df = _rows_to_df(env._db.original_state())
+
+            # Hard task stats
+            if task_id.endswith("_hard") and hasattr(env, "_db") and env._db:
+                hard_issues_total = env._db.total_issues
+                hard_table        = table
+                hard_columns      = visible
+
+            tbl_note = f"Tables: {all_tables}" if len(all_tables) > 1 else f"Table: {table}"
+            _log(f"  {tbl_note}  ({t_info.get('row_count','?')} rows, {len(visible)} cols)")
+            yield emit()
 
             # Phase 1: Inspect
             r = _do({"action_type": "inspect", "table": table})
-            _log(f"  [step {step:2d}] inspect         reward={r:+.2f}")
-            yield _current_log(), summary_md, None
+            _log(f"  [{step:02d}] inspect                reward={r:+.2f}")
+            yield emit()
+
+            # Phase 1b: Multi-table — select + join
+            if len(all_tables) > 1:
+                r = _do({"action_type": "select_tables", "tables": all_tables})
+                _log(f"  [{step:02d}] select_tables({all_tables})  reward={r:+.2f}")
+                yield emit()
+                if len(all_tables) >= 2 and not done:
+                    t2_name = all_tables[1]
+                    t2_cols = obs.tables_summary.get(t2_name, {}).get("columns", [])
+                    shared  = [c for c in visible if c in t2_cols
+                               and c not in ("id", "_source_format")]
+                    jkey    = shared[0] if shared else (visible[0] if visible else None)
+                    if jkey:
+                        r = _do({"action_type": "join_tables",
+                                 "table": table, "table2": t2_name, "key": jkey})
+                        _log(f"  [{step:02d}] join({table},{t2_name},key={jkey})  reward={r:+.2f}")
+                        yield emit()
 
             # Phase 2: Profile + bulk-fix
             for col in visible[:12]:
@@ -272,44 +637,88 @@ def run_agent_streaming(
                     p  = obs.query_result[0]
                     nc = p.get("null_count", 0)
                     dt = p.get("dtype", "?")
-                    _log(f"  [step {step:2d}] profile({col:<12}) type={dt}  nulls={nc}  reward={r:+.2f}")
-                    yield _current_log(), summary_md, None
-
+                    _log(f"  [{step:02d}] profile({col:<12})  type={dt}  nulls={nc}  reward={r:+.2f}")
+                    yield emit()
                     if nc > 0 and not done:
-                        fill     = (p.get("median") or p.get("mean") or 0) if dt in ("int", "float") else (p.get("mode") or "Unknown")
-                        strategy = "median" if dt in ("int", "float") else "mode"
+                        fill     = (p.get("median") or p.get("mean") or 0) if dt in ("int","float") \
+                                   else (p.get("mode") or "Unknown")
+                        strategy = "median" if dt in ("int","float") else "mode"
                         r2 = _do({
                             "action_type": "fix_column", "table": table,
                             "column": col, "value": fill,
-                            "reason": f"bulk fix {nc} nulls, {strategy}={fill}",
+                            "reason": f"bulk fix {nc} nulls in {col}, {strategy}={fill}",
                         })
-                        _log(f"  [step {step:2d}] fix_column({col:<12}, {str(fill)[:8]!r:<10})  reward={r2:+.2f}")
+                        _log(f"  [{step:02d}] fix_column({col:<12}, {str(fill)[:8]!r})  reward={r2:+.2f}")
                         if r2 > 0:
-                            fixed_cols.append(col)
-                        yield _current_log(), summary_md, None
+                            cols_fixed_this_task.append(col)
+                        yield emit()
                 else:
-                    _log(f"  [step {step:2d}] profile({col:<12})  reward={r:+.2f}")
-                    yield _current_log(), summary_md, None
+                    _log(f"  [{step:02d}] profile({col:<12})  reward={r:+.2f}")
+                    yield emit()
 
             # Phase 3: Validate
             if not done:
                 r  = _do({"action_type": "validate"})
                 fb = (obs.last_feedback if obs else "")[:100]
-                _log(f"  [step {step:2d}] validate         reward={r:+.2f}  {fb}")
-                yield _current_log(), summary_md, None
-
-            # Phase 4: Export (hard tasks produce a downloadable file)
+                _log(f"  [{step:02d}] validate               reward={r:+.2f}  {fb}")
+                yield emit()
+
+            # Phase 4: Business query SQL (run before export so DB is still live)
+            if not done and task_id.endswith("_hard"):
+                # Determine SQL to run
+                run_sql = biz_sql.strip() if biz_sql.strip() else None
+                if run_sql is None and intent == "business_query" and user_requirement.strip():
+                    run_sql = _nl_to_sql(user_requirement, table, visible)
+                    if run_sql:
+                        auto_sql_used = run_sql
+                        _log(f"  [auto-SQL generated from question]")
+                        _log(f"  → {run_sql[:80]}{'…' if len(run_sql) > 80 else ''}")
+                        yield emit()
+
+                if run_sql and run_sql.upper().startswith("SELECT"):
+                    r = _do({"action_type": "run_sql", "sql": run_sql[:500]})
+                    if obs and obs.query_result:
+                        query_df = pd.DataFrame(obs.query_result)
+                        _log(f"  [{step:02d}] run_sql → {len(obs.query_result)} rows  reward={r:+.2f}")
+                    else:
+                        _log(f"  [{step:02d}] run_sql → no results  reward={r:+.2f}")
+                    yield emit()
+
+            # Phase 5: Export
             if not done:
+                # Record issues remaining before export (Hard task only)
+                if task_id.endswith("_hard") and hasattr(env, "_db") and env._db:
+                    hard_issues_remain  = env._db.issues_remaining()
+                    hard_columns_fixed  = cols_fixed_this_task[:]
+
                 r  = _do({"action_type": "export"})
-                fb = (obs.last_feedback if obs else "")[:120]
-                _log(f"  [step {step:2d}] export           reward={r:+.2f}")
+                fb = (obs.last_feedback if obs else "")[:130]
+                _log(f"  [{step:02d}] export                 reward={r:+.2f}")
                 _log(f"         {fb}")
                 m = re.search(r"[Gg]rader\s+score\s*=?\s*(\d+\.\d+)", fb)
                 if m:
                     score = float(m.group(1))
-                if task_id.endswith("_hard") and hasattr(env, "_export_result") and env._export_result:
-                    download_path = env._export_result.get("filepath")
-                yield _current_log(), summary_md, download_path
+
+                # Capture after-state and download from Hard task
+                if task_id.endswith("_hard") and hasattr(env, "_db") and env._db:
+                    after_df = _rows_to_df(env._db.current_state())
+                    if hasattr(env, "_export_result") and env._export_result:
+                        download_path = env._export_result.get("filepath")
+
+                    # Build final answer now that we have all data
+                    final_answer = _generate_final_answer(
+                        intent         = intent,
+                        user_requirement = user_requirement,
+                        issues_total   = hard_issues_total,
+                        issues_remaining = hard_issues_remain,
+                        columns_fixed  = hard_columns_fixed,
+                        query_df       = query_df,
+                        auto_sql       = auto_sql_used,
+                        table          = hard_table or table,
+                        columns        = hard_columns or visible,
+                    )
+
+                yield emit()
 
         except Exception as exc:
             _log(f"  [ERROR] {_friendly_error(exc, source)}")
@@ -319,18 +728,16 @@ def run_agent_streaming(
             except Exception:
                 pass
 
-        total_r = sum(rewards)
         success = score >= 0.50
-        _log(f"  Score: {score:.2f}  Total reward: {total_r:+.3f}  "
-             f"{'SUCCESS' if success else 'PARTIAL'}  Steps: {step}")
+        _log(f"  Score: {score:.2f}  {'SUCCESS ✓' if success else 'PARTIAL'}  Steps: {step}")
         _log("")
         all_scores.append((task_id, score))
-        yield _current_log(), summary_md, download_path
+        yield emit()
 
     # Final summary table
     avg = sum(s for _, s in all_scores) / len(all_scores) if all_scores else 0.0
     rows_md = "\n".join(
-        f"| `{tid}` | {s:.2f} | {'Pass' if s >= 0.50 else 'Fail'} |"
+        f"| `{tid}` | {s:.2f} | {'Pass ✓' if s >= 0.50 else 'Fail ✗'} |"
         for tid, s in all_scores
     )
     summary_md = f"""## Results — Intent: `{intent}`
@@ -343,25 +750,31 @@ def run_agent_streaming(
 
 | Detail | Value |
 |--------|-------|
-| Dataset | `{source}` |
-| Output format | `{output_format}` |
+| Dataset | `{src_label}` |
+| Format | `{output_format}` |
 | Tasks run | {len(all_scores)} (easy → medium → hard) |
-{"| **Download** | cleaned file available below |" if download_path else ""}
+{"| Download | cleaned file ready ↓ |" if download_path else ""}
+{"| Answer | results shown ↓ |" if not query_df.empty else ""}
 """
-    yield _current_log(), summary_md, download_path
+    yield emit()
 
 
 # ---------------------------------------------------------------------------
-# Build the Gradio Blocks app
+# Build Gradio UI
 # ---------------------------------------------------------------------------
 
+def _intent_req_update(intent: str):
+    cfg = _REQ_CONFIG.get(intent, _REQ_CONFIG["visualization"])
+    return gr.update(label=cfg["label"], placeholder=cfg["placeholder"])
+
+
 def create_ui() -> gr.Blocks:
     with gr.Blocks(title="SQLSherlock — AI Data Quality Detective") as demo:
 
         gr.Markdown(
             "# SQLSherlock — AI Data Quality Detective\n"
-            "Select your **cleaning intent** and let the agent clean your dataset "
-            "across **3 escalating difficulty levels** — Easy, Medium, then Hard."
+            "Describe what you want to do with your data — "
+            "the agent cleans it across **3 difficulty levels** and answers your question."
         )
 
         # ==================================================================
@@ -369,25 +782,23 @@ def create_ui() -> gr.Blocks:
         # ==================================================================
         with gr.Tab("Data Quality Scanner"):
             gr.Markdown(
-                "Provide a dataset, then click **Scan** to see a full quality report."
-                "\n\n> If both file and name are filled, the uploaded file takes priority."
+                "Scan any dataset for data quality issues before running the agent.\n\n"
+                "> If both a file and a name are provided, the uploaded file takes priority."
             )
             with gr.Row():
                 with gr.Column(scale=1):
-                    gr.Markdown("#### Option A — Upload a file")
+                    gr.Markdown("#### Upload a file")
                     scan_file = gr.File(
                         label="CSV / JSON / JSONL / Parquet / Excel",
                         file_types=[".csv", ".json", ".jsonl", ".parquet", ".xlsx"],
                         type="filepath",
                     )
-                    gr.Markdown("#### Option B — HuggingFace dataset")
+                    gr.Markdown("#### OR — HuggingFace dataset")
                     scan_hf = gr.Textbox(
-                        label="Dataset name",
+                        label="Dataset name", value="phihung/titanic",
                         placeholder="e.g. phihung/titanic",
-                        value="phihung/titanic",
                     )
-                    gr.Examples(examples=_EXAMPLE_HF_DATASETS, inputs=[scan_hf],
-                                label="Quick examples")
+                    gr.Examples(examples=_EXAMPLE_HF, inputs=[scan_hf], label="Quick examples")
                     scan_btn = gr.Button("Scan Dataset", variant="primary", size="lg")
 
                 with gr.Column(scale=2):
@@ -418,38 +829,63 @@ def create_ui() -> gr.Blocks:
             )
 
         # ==================================================================
-        # Tab 2: Agent Demo — intent → 3 difficulty levels
+        # Tab 2: Agent Demo
         # ==================================================================
         with gr.Tab("Agent Demo"):
             gr.Markdown(
-                "**Choose one cleaning intent** — the agent will automatically run "
-                "**Easy → Medium → Hard** tasks for that intent.\n\n"
-                "The cleaned file from the Hard task is available for download."
+                "**Describe what you want** — the agent cleans your data "
+                "and answers your question or prepares data for your specific use case."
             )
 
             with gr.Row():
-                # --- Dataset input ---
+                # --- Dataset ---
                 with gr.Column(scale=2):
-                    gr.Markdown("#### Dataset")
-                    agent_file = gr.File(
-                        label="Upload file (CSV / JSON / JSONL / Parquet / XLSX)",
-                        file_types=[".csv", ".json", ".jsonl", ".parquet", ".xlsx"],
-                        type="filepath",
+                    gr.Markdown("#### Dataset (upload up to 3 tables)")
+                    with gr.Row():
+                        agent_file1 = gr.File(
+                            label="Table 1 (primary)",
+                            file_types=[".csv", ".json", ".jsonl", ".parquet", ".xlsx"],
+                            type="filepath",
+                        )
+                        agent_file2 = gr.File(
+                            label="Table 2 (optional)",
+                            file_types=[".csv", ".json", ".jsonl", ".parquet", ".xlsx"],
+                            type="filepath",
+                        )
+                        agent_file3 = gr.File(
+                            label="Table 3 (optional)",
+                            file_types=[".csv", ".json", ".jsonl", ".parquet", ".xlsx"],
+                            type="filepath",
+                        )
+                    gr.Markdown(
+                        "> Upload **2–3 files** for multi-table mode — they are merged into "
+                        "one XLSX and the agent will inspect all schemas and attempt joins.\n\n"
+                        "> OR enter a **HuggingFace dataset** name below (used only when no file is uploaded)."
                     )
                     agent_hf = gr.Textbox(
-                        label="OR enter HuggingFace dataset name",
+                        label="HuggingFace dataset name",
                         value="phihung/titanic",
                         placeholder="phihung/titanic",
                     )
 
-                # --- Configuration ---
-                with gr.Column(scale=1):
+                # --- Intent + Requirement ---
+                with gr.Column(scale=2):
                     gr.Markdown("#### Cleaning Intent")
                     agent_intent = gr.Radio(
                         choices=INTENT_CHOICES,
                         label="What will you use this data for?",
                         value="visualization",
                     )
+                    # Dynamic text area — label and placeholder change with intent
+                    requirement_input = gr.Textbox(
+                        label=_REQ_CONFIG["visualization"]["label"],
+                        placeholder=_REQ_CONFIG["visualization"]["placeholder"],
+                        lines=3,
+                        max_lines=6,
+                    )
+
+                # --- Format ---
+                with gr.Column(scale=1):
                     gr.Markdown("#### Output Format")
                     agent_format = gr.Radio(
                         choices=FORMAT_CHOICES,
@@ -457,26 +893,48 @@ def create_ui() -> gr.Blocks:
                         value="csv",
                     )
 
+            # Update requirement box when intent changes
+            agent_intent.change(
+                fn=_intent_req_update,
+                inputs=[agent_intent],
+                outputs=[requirement_input],
+            )
+
+            # Advanced SQL box
+            with gr.Accordion("Advanced: Direct SQL Query (optional)", open=False):
+                gr.Markdown(
+                    "Enter a `SELECT` query to run on the cleaned dataset after the Hard task. "
+                    "If left empty and intent is **Business Analytics**, an SQL query is "
+                    "auto-generated from your requirement text above.\n\n"
+                    "Use the table name shown in the step log (e.g. `dataset`, `titanic`)."
+                )
+                biz_sql_input = gr.Textbox(
+                    label="SQL Query",
+                    placeholder=_SQL_PLACEHOLDER,
+                    lines=5,
+                    max_lines=12,
+                )
+
             gr.Markdown(
-                "> The agent will run 3 tasks for the selected intent:\n"
-                "> - **Easy** — basic cleaning (nulls, type errors, whitespace, categories)\n"
-                "> - **Medium** — + constraints, outliers\n"
-                "> - **Hard** — + duplicates, FK violations, trap detection"
+                "> **Scoring cascade — Easy matters most:**\n"
+                "> Easy `res×0.80 + val×0.20` → Medium `easy×0.70 + advanced×0.30` "
+                "→ Hard `medium×0.70 + FK×0.30`"
             )
 
             gr.Examples(
                 examples=_AGENT_EXAMPLES,
-                inputs=[agent_hf, agent_intent, agent_format],
+                inputs=[agent_hf, agent_intent, requirement_input, agent_format, biz_sql_input],
                 label="Quick examples",
             )
 
             run_btn = gr.Button("Run Agent (3 Tasks)", variant="primary", size="lg")
 
+            # --- Primary outputs ---
             with gr.Row():
                 with gr.Column(scale=3):
                     agent_log = gr.Textbox(
                         label="Live Step Log",
-                        lines=30, max_lines=120,
+                        lines=28, max_lines=120,
                         interactive=False,
                     )
                 with gr.Column(scale=2):
@@ -484,22 +942,53 @@ def create_ui() -> gr.Blocks:
                         value="_Results appear here after the agent finishes._"
                     )
 
+            schema_out = gr.Markdown(value="", label="Schema (multi-table)")
+
             gr.Markdown("#### Download Cleaned File (Hard task output)")
             download_out = gr.File(
-                label="Cleaned dataset — available after the Hard task completes",
+                label="Cleaned dataset",
                 interactive=False,
             )
 
-            def _run(file_path, hf_name, intent, output_format):
-                for log, results, dl in run_agent_streaming(
-                    file_path, hf_name, intent, output_format
+            # --- Before / After ---
+            gr.Markdown("#### Before / After Comparison")
+            with gr.Row():
+                before_tbl = gr.Dataframe(
+                    label="Original Data (before agent)",
+                    interactive=False, wrap=True,
+                )
+                after_tbl = gr.Dataframe(
+                    label="Cleaned Data (after Hard task)",
+                    interactive=False, wrap=True,
+                )
+
+            # --- Final Answer ---
+            gr.Markdown("#### Final Answer")
+            final_answer_out = gr.Markdown(
+                value="_Final answer appears here after all 3 tasks complete._"
+            )
+
+            # --- Business query result table ---
+            gr.Markdown("#### Query Results")
+            query_result_tbl = gr.Dataframe(
+                label="SQL query output (on cleaned data)",
+                interactive=False, wrap=True,
+            )
+
+            def _run(f1, f2, f3, hf, intent, fmt, req, sql):
+                for log, summary, dl, bdf, adf, schema, qdf, answer in run_agent_streaming(
+                    f1, f2, f3, hf, intent, fmt, req, sql
                 ):
-                    yield log, results, dl
+                    yield log, summary, dl, bdf, adf, schema, qdf, answer
 
             run_btn.click(
                 fn=_run,
-                inputs=[agent_file, agent_hf, agent_intent, agent_format],
-                outputs=[agent_log, agent_results, download_out],
+                inputs=[agent_file1, agent_file2, agent_file3,
+                        agent_hf, agent_intent, agent_format,
+                        requirement_input, biz_sql_input],
+                outputs=[agent_log, agent_results, download_out,
+                         before_tbl, after_tbl, schema_out,
+                         query_result_tbl, final_answer_out],
                 show_progress="minimal",
             )
 
@@ -508,62 +997,39 @@ def create_ui() -> gr.Blocks:
         # ==================================================================
         with gr.Tab("About"):
             gr.Markdown("""
-## SQLSherlock Task Architecture
-
-Each **cleaning intent** has **3 difficulty levels** — the agent always
-runs Easy → Medium → Hard for the chosen intent.
-
-### Intents & Their 3 Tasks
-
-| Intent | Easy | Medium | Hard |
-|--------|------|--------|------|
-| **Visualization** | `viz_easy` | `viz_medium` | `viz_hard` |
-| **ML Training** | `ml_easy` | `ml_medium` | `ml_hard` |
-| **Business Analytics** | `bq_easy` | `bq_medium` | `bq_hard` |
-
-### Issue Coverage per Difficulty
-
-| Difficulty | Issues Covered | Max Steps |
-|------------|----------------|-----------|
-| **Easy** | nulls, type errors, whitespace, inconsistent categories | 30 |
-| **Medium** | + constraint violations, statistical outliers (z > 5) | 40 |
-| **Hard** | + duplicate rows, FK violations — plus the **trap** | 50 |
+## How It Works
 
-### Scoring Formulas
+1. **Describe your goal** — dashboard, model, or business question
+2. **Agent cleans Easy → Medium → Hard** for the selected intent
+3. **Final answer** — cleaning summary + SQL results + narrative
 
-**Easy:**
-```
-score = resolution(null+type+whitespace+category) × 0.70 + validation × 0.30
-```
+### Scoring (Easy → Medium → Hard cascade)
 
-**Medium:**
-```
-score = easy_score × 0.40 + avg(constraint + outlier) × 0.60
-```
+| Task | Formula | Key weight |
+|------|---------|------------|
+| **Easy** | `resolution × 0.80 + validation × 0.20` | Nulls/types/whitespace/categories |
+| **Medium** | `easy × 0.70 + advanced × 0.30` | Easy = 70% of medium |
+| **Hard** | `medium × 0.70 + FK × 0.30` | Medium = 70% of hard |
 
-**Hard:**
-```
-score = medium_score × 0.50 + fk_resolved × 0.50 + reasoning_bonus − trap_penalty
-```
+### Multiple Tables
 
-### The Trap (Hard tasks only)
+Upload **2–3 files** — each becomes a table (sheet in a merged XLSX).
+The agent will `select_tables`, inspect schemas, and attempt `join_tables`.
 
-A numeric value is doubled — it *looks* like an outlier but z < 3.
-Touching it costs **−0.40**. Always verify z-scores before fixing numeric values.
+### Business Query Answering
 
----
+- Enter a **plain English question** in the requirement box
+- The agent auto-generates SQL from keywords (group-by, average, top-N, etc.)
+- Or provide exact SQL in the Advanced SQL field
+- Results appear in the **Query Results** table with a narrative
 
-### API Endpoints
+### Intent-Specific Cleaning Focus
 
-| Method | Path | Description |
-|--------|------|-------------|
-| WebSocket | `/ws` | RL episode session |
-| POST | `/reset` | Start episode (`task_id`, `intent`, `output_format`) |
-| POST | `/step` | Execute one action |
-| GET | `/health` | Health check + task list |
-| GET | `/tasks` | All 9 task definitions |
-| POST | `/upload_dataset` | Upload CSV/JSON/Parquet/XLSX |
-| GET | `/download/{id}` | Download cleaned output |
+| Intent | Primary Issues | Output |
+|--------|---------------|--------|
+| **Visualization** | Nulls, whitespace, categories | Chart-ready columns list |
+| **ML Training** | Nulls, type errors, outliers, constraints | Feature/target column list |
+| **Business Analytics** | All types + FK + duplicates | SQL answer + narrative |
 """)
 
     return demo

tests/conftest.py

@@ -127,10 +127,10 @@ def clean_profile():
 
 @pytest.fixture
 def db_task1():
-    """DatabaseEngine for task1 loaded from raw CSV text."""
+    """DatabaseEngine for easy task (viz_easy) loaded from raw CSV text."""
     from server.database import DatabaseEngine
     db = DatabaseEngine(
-        task_id="task1_null_and_types",
+        task_id="viz_easy",
         seed=42,
         dataset_source=RAW_CSV_TEXT,
         max_rows=50,
@@ -140,10 +140,10 @@ def db_task1():
 
 @pytest.fixture
 def db_task2():
-    """DatabaseEngine for task2 loaded from raw CSV text."""
+    """DatabaseEngine for medium task (ml_medium) loaded from raw CSV text."""
     from server.database import DatabaseEngine
     db = DatabaseEngine(
-        task_id="task2_constraints_and_fk",
+        task_id="ml_medium",
         seed=42,
         dataset_source=RAW_CSV_TEXT,
         max_rows=50,
@@ -153,10 +153,10 @@ def db_task2():
 
 @pytest.fixture
 def db_task3():
-    """DatabaseEngine for task3 loaded from raw CSV text."""
+    """DatabaseEngine for hard task (bq_hard) loaded from raw CSV text."""
     from server.database import DatabaseEngine
     db = DatabaseEngine(
-        task_id="task3_full_audit_with_trap",
+        task_id="bq_hard",
         seed=42,
         dataset_source=RAW_CSV_TEXT,
         max_rows=50,
@@ -170,7 +170,7 @@ def db_task3():
 
 @pytest.fixture
 def task1_issues(dirty_conn, dirty_profile):
-    """Issues detected for task1 on the dirty dataset."""
+    """Issues detected for viz_easy task on the dirty dataset."""
     from server.issue_detector import detect_issues
     import copy
     records = copy.deepcopy(DIRTY_RECORDS)
@@ -178,14 +178,14 @@ def task1_issues(dirty_conn, dirty_profile):
         conn=dirty_conn,
         profile=dirty_profile,
         records=records,
-        task_id="task1_null_and_types",
+        task_id="viz_easy",
         seed=42,
     )
 
 
 @pytest.fixture
 def task3_issues(dirty_conn, dirty_profile):
-    """Issues detected for task3 on the dirty dataset."""
+    """Issues detected for bq_hard task on the dirty dataset."""
     from server.issue_detector import detect_issues
     import copy
     records = copy.deepcopy(DIRTY_RECORDS)
@@ -193,6 +193,54 @@ def task3_issues(dirty_conn, dirty_profile):
         conn=dirty_conn,
         profile=dirty_profile,
         records=records,
-        task_id="task3_full_audit_with_trap",
+        task_id="bq_hard",
         seed=42,
     )
+
+
+# ---------------------------------------------------------------------------
+# Multi-table XLSX fixture
+# ---------------------------------------------------------------------------
+
+@pytest.fixture
+def multi_table_xlsx_path():
+    """Create a temp XLSX with two sheets for multi-table testing.
+
+    Sheet 1 — 'passengers': 10 rows with null/type/constraint issues.
+    Sheet 2 — 'classes':     3 rows (lookup table, clean).
+    Both sheets share the 'id' column so join_tables can be tested.
+    """
+    import tempfile
+    import pandas as pd
+
+    df_passengers = pd.DataFrame([
+        {"id": 1,  "name": "Alice",  "age": None,    "fare": 10.50, "survived": 1},
+        {"id": 2,  "name": "Bob",    "age": 25,      "fare": 7.25,  "survived": 0},
+        {"id": 3,  "name": "Carol",  "age": "FORTY", "fare": 15.00, "survived": 1},
+        {"id": 4,  "name": "Dave",   "age": -5,      "fare": 8.00,  "survived": 0},
+        {"id": 5,  "name": "Eve",    "age": 28,      "fare": 12.00, "survived": 1},
+        {"id": 6,  "name": "Frank",  "age": 45,      "fare": 9.75,  "survived": 0},
+        {"id": 7,  "name": "Grace",  "age": 33,      "fare": 11.50, "survived": 1},
+        {"id": 8,  "name": "Heidi",  "age": 29,      "fare": 6.50,  "survived": 0},
+        {"id": 9,  "name": "Ivan",   "age": 38,      "fare": 13.25, "survived": 1},
+        {"id": 10, "name": "Judy",   "age": 22,      "fare": 5.00,  "survived": 0},
+    ])
+
+    df_classes = pd.DataFrame([
+        {"id": 1, "class_name": "First",   "min_fare": 30.0},
+        {"id": 2, "class_name": "Second",  "min_fare": 10.0},
+        {"id": 3, "class_name": "Third",   "min_fare": 5.0},
+        {"id": 4, "class_name": "Fourth",  "min_fare": 3.0},
+        {"id": 5, "class_name": "Fifth",   "min_fare": 2.0},
+    ])
+
+    tmp = tempfile.NamedTemporaryFile(suffix=".xlsx", delete=False)
+    tmp.close()
+
+    with pd.ExcelWriter(tmp.name, engine="openpyxl") as writer:
+        df_passengers.to_excel(writer, sheet_name="passengers", index=False)
+        df_classes.to_excel(writer, sheet_name="classes", index=False)
+
+    yield tmp.name
+
+    os.unlink(tmp.name)

tests/test_environment.py

@@ -588,3 +588,201 @@ class TestJoinTables:
         ))
         state = env_task1.get_state()
         assert state.joins_performed == 1
+
+
+# ---------------------------------------------------------------------------
+# Multi-table input (2-sheet XLSX)
+# ---------------------------------------------------------------------------
+
+class TestMultiTableInput:
+    """Tests for loading and operating on a 2-sheet XLSX dataset."""
+
+    @pytest.fixture
+    def env_multi(self, env, multi_table_xlsx_path):
+        """Reset env with the 2-sheet XLSX — both tables are available."""
+        env.reset(dataset=multi_table_xlsx_path, task_id="viz_easy")
+        return env
+
+    # ------------------------------------------------------------------
+    # Loading
+    # ------------------------------------------------------------------
+
+    def test_two_tables_loaded(self, env_multi):
+        """Both XLSX sheets must appear as separate SQLite tables."""
+        tables = list(env_multi._db.table_names())
+        assert len(tables) == 2, f"Expected 2 tables, got {tables}"
+
+    def test_passengers_table_present(self, env_multi):
+        tables = list(env_multi._db.table_names())
+        assert "passengers" in tables
+
+    def test_classes_table_present(self, env_multi):
+        tables = list(env_multi._db.table_names())
+        assert "classes" in tables
+
+    def test_passengers_row_count(self, env_multi):
+        rows = env_multi._db.rows("passengers")
+        assert len(rows) == 10
+
+    def test_classes_row_count(self, env_multi):
+        rows = env_multi._db.rows("classes")
+        assert len(rows) == 5
+
+    # ------------------------------------------------------------------
+    # select_tables — reward should be positive for ≥2 tables
+    # ------------------------------------------------------------------
+
+    def test_select_tables_rewards_when_multi_table(self, env_multi):
+        """select_tables earns +0.02 when the dataset has ≥2 tables."""
+        obs, reward, done, _ = _step(env_multi,
+            SQLSherlockAction(action_type="select_tables",
+                              tables=["passengers", "classes"])
+        )
+        assert reward > 0, f"Expected positive reward, got {reward}"
+
+    def test_select_tables_updates_state_with_both_tables(self, env_multi):
+        env_multi.step(SQLSherlockAction(
+            action_type="select_tables",
+            tables=["passengers", "classes"],
+        ))
+        state = env_multi.get_state()
+        assert "passengers" in state.tables_selected
+        assert "classes" in state.tables_selected
+
+    # ------------------------------------------------------------------
+    # inspect second table
+    # ------------------------------------------------------------------
+
+    def test_inspect_classes_table(self, env_multi):
+        obs, reward, done, _ = _step(env_multi,
+            SQLSherlockAction(action_type="inspect", table="classes")
+        )
+        assert obs.query_result is not None
+        assert len(obs.query_result) == 5
+
+    def test_inspect_classes_has_class_name_column(self, env_multi):
+        obs, _, _, _ = _step(env_multi,
+            SQLSherlockAction(action_type="inspect", table="classes")
+        )
+        assert "class_name" in obs.query_result[0]
+
+    # ------------------------------------------------------------------
+    # profile_column on second table
+    # ------------------------------------------------------------------
+
+    def test_profile_column_on_classes_table(self, env_multi):
+        obs, reward, done, _ = _step(env_multi,
+            SQLSherlockAction(action_type="profile_column",
+                              table="classes", column="min_fare")
+        )
+        assert obs.query_result is not None
+        profile = obs.query_result[0]
+        assert "mean" in profile
+        assert "median" in profile
+
+    def test_profile_column_classes_min_fare_mean(self, env_multi):
+        """min_fare values are 30.0, 10.0, 5.0, 3.0, 2.0 → mean = 10.0."""
+        obs, _, _, _ = _step(env_multi,
+            SQLSherlockAction(action_type="profile_column",
+                              table="classes", column="min_fare")
+        )
+        profile = obs.query_result[0]
+        assert abs(profile["mean"] - 10.0) < 1.0
+
+    # ------------------------------------------------------------------
+    # join_tables between the two real tables
+    # ------------------------------------------------------------------
+
+    def test_join_tables_valid_gives_positive_reward(self, env_multi):
+        """join_tables on a shared key column earns +0.20."""
+        obs, reward, done, _ = _step(env_multi,
+            SQLSherlockAction(
+                action_type="join_tables",
+                table="passengers",
+                table2="classes",
+                key="id",
+            )
+        )
+        assert reward > 0, f"Expected positive reward for valid join, got {reward}"
+
+    def test_join_tables_valid_increments_joins_performed(self, env_multi):
+        env_multi.step(SQLSherlockAction(
+            action_type="join_tables",
+            table="passengers",
+            table2="classes",
+            key="id",
+        ))
+        state = env_multi.get_state()
+        assert state.joins_performed == 1
+
+    def test_join_tables_result_has_columns_from_both_tables(self, env_multi):
+        obs, _, _, _ = _step(env_multi,
+            SQLSherlockAction(
+                action_type="join_tables",
+                table="passengers",
+                table2="classes",
+                key="id",
+            )
+        )
+        # Join result should appear in query_result or last_feedback
+        # At minimum the step must not error
+        assert "error" not in (obs.last_feedback or "").lower() or obs.query_result is not None
+
+    # ------------------------------------------------------------------
+    # run_sql across two tables
+    # ------------------------------------------------------------------
+
+    def test_run_sql_on_second_table(self, env_multi):
+        obs, reward, done, _ = _step(env_multi,
+            SQLSherlockAction(
+                action_type="run_sql",
+                sql='SELECT class_name, min_fare FROM "classes" ORDER BY min_fare DESC',
+            )
+        )
+        assert obs.query_result is not None
+        assert len(obs.query_result) == 5
+        # Should be ordered descending — First class has highest fare
+        assert obs.query_result[0]["class_name"] == "First"
+
+    def test_run_sql_cross_table_join(self, env_multi):
+        obs, _, _, _ = _step(env_multi,
+            SQLSherlockAction(
+                action_type="run_sql",
+                sql=(
+                    'SELECT p.name, c.class_name '
+                    'FROM "passengers" p '
+                    'LEFT JOIN "classes" c ON p.id = c.id '
+                    'LIMIT 3'
+                ),
+            )
+        )
+        assert obs.query_result is not None
+        assert len(obs.query_result) == 3
+
+    # ------------------------------------------------------------------
+    # fix_column on second table
+    # ------------------------------------------------------------------
+
+    def test_fix_column_no_issues_on_classes(self, env_multi):
+        """Applying fix_column to a clean column gives a negative reward (false positive)."""
+        obs, reward, done, _ = _step(env_multi,
+            SQLSherlockAction(
+                action_type="fix_column",
+                table="classes",
+                column="min_fare",
+                value=10.0,
+                reason="test unnecessary fix",
+            )
+        )
+        # Clean column → negative reward (−0.10 for fix_column with no issues)
+        assert reward <= 0, f"Expected ≤ 0 reward for clean column fix, got {reward}"
+
+    # ------------------------------------------------------------------
+    # Multi-table episode reaches done correctly
+    # ------------------------------------------------------------------
+
+    def test_multi_table_submit_ends_episode(self, env_multi):
+        _, _, done, _ = _step(env_multi,
+            SQLSherlockAction(action_type="submit")
+        )
+        assert done is True

tests/test_graders.py

@@ -111,7 +111,7 @@ class TestZeroChangeGuard:
             db=db_task1,
             cleaned_rows=dirty,
             removed_ids=[],
-            task_id="task1_null_and_types",
+            task_id="viz_easy",
             validation_was_called=False,
         )
         assert score == 0.0
@@ -136,7 +136,7 @@ class TestTask1Grader:
             db=db_task1,
             cleaned_rows=cleaned,
             removed_ids=removed,
-            task_id="task1_null_and_types",
+            task_id="viz_easy",
             validation_was_called=True,
         )
         assert score >= 0.60, f"Expected >= 0.60 after full fix, got {score}"
@@ -147,7 +147,7 @@ class TestTask1Grader:
             db=db_task1,
             cleaned_rows=dirty,
             removed_ids=[],
-            task_id="task1_null_and_types",
+            task_id="viz_easy",
             validation_was_called=False,
         )
         assert score == 0.0
@@ -158,15 +158,15 @@ class TestTask1Grader:
             db=db_task1,
             cleaned_rows=cleaned,
             removed_ids=[],
-            task_id="task1_null_and_types",
+            task_id="viz_easy",
             validation_was_called=True,
         )
         assert 0.0 <= score <= 1.0
 
     def test_no_validate_penalty(self, db_task1):
         cleaned = _apply_all_fixes(db_task1)
-        score_with    = graders.grade(db_task1, cleaned, [], "task1_null_and_types", True)
-        score_without = graders.grade(db_task1, cleaned, [], "task1_null_and_types", False)
+        score_with    = graders.grade(db_task1, cleaned, [], "viz_easy", True)
+        score_without = graders.grade(db_task1, cleaned, [], "viz_easy", False)
         assert score_with >= score_without
 
     def test_false_positive_reduces_score(self, db_task1):
@@ -177,8 +177,8 @@ class TestTask1Grader:
             if row.get("survived") is not None:
                 row["survived"] = 99   # not an issue
                 break
-        score_fp  = graders.grade(db_task1, clean_copy, [], "task1_null_and_types", True)
-        score_ok  = graders.grade(db_task1, cleaned,    [], "task1_null_and_types", True)
+        score_fp  = graders.grade(db_task1, clean_copy, [], "viz_easy", True)
+        score_ok  = graders.grade(db_task1, cleaned,    [], "viz_easy", True)
         assert score_fp <= score_ok
 
 
@@ -197,7 +197,7 @@ class TestTask2Grader:
             db=db_task2,
             cleaned_rows=cleaned,
             removed_ids=removed,
-            task_id="task2_constraints_and_fk",
+            task_id="ml_medium",
             validation_was_called=True,
         )
         assert score >= 0.50, f"Expected >= 0.50 after full fix, got {score}"
@@ -208,7 +208,7 @@ class TestTask2Grader:
             db=db_task2,
             cleaned_rows=cleaned,
             removed_ids=[],
-            task_id="task2_constraints_and_fk",
+            task_id="ml_medium",
             validation_was_called=True,
         )
         assert 0.0 <= score <= 1.0
@@ -217,8 +217,8 @@ class TestTask2Grader:
         """task2 weight means full fix may score differently — both must be in range."""
         c1 = _apply_all_fixes(db_task1)
         c2 = _apply_all_fixes(db_task2)
-        s1 = graders.grade(db_task1, c1, [], "task1_null_and_types",     True)
-        s2 = graders.grade(db_task2, c2, [], "task2_constraints_and_fk", True)
+        s1 = graders.grade(db_task1, c1, [], "viz_easy",     True)
+        s2 = graders.grade(db_task2, c2, [], "ml_medium", True)
         assert 0.0 <= s1 <= 1.0
         assert 0.0 <= s2 <= 1.0
 
@@ -234,7 +234,7 @@ class TestTask3Grader:
             db=db_task3,
             cleaned_rows=cleaned,
             removed_ids=[],
-            task_id="task3_full_audit_with_trap",
+            task_id="bq_hard",
             validation_was_called=True,
         )
         assert 0.0 <= score <= 1.0
@@ -256,11 +256,11 @@ class TestTask3Grader:
 
         score_untouched = graders.grade(
             db_task3, cleaned_no_touch, [],
-            "task3_full_audit_with_trap", True,
+            "bq_hard", True,
         )
         score_touched = graders.grade(
             db_task3, cleaned_touched, [],
-            "task3_full_audit_with_trap", True,
+            "bq_hard", True,
         )
         assert score_touched < score_untouched or score_touched <= score_untouched
 
@@ -282,7 +282,7 @@ class TestTask3Grader:
         cleaned = _apply_all_fixes(db_task3)
         score_with_reason = graders.grade(
             db_task3, cleaned, [],
-            "task3_full_audit_with_trap", True,
+            "bq_hard", True,
         )
         assert score_with_reason >= 0.0
 
@@ -351,4 +351,4 @@ class TestFalsePositivePenalty:
         penalty = _false_positive_penalty(
             db_task1, tampered, [], db_task1.pk_col, db_task1.primary_table
         )
-        assert penalty <= 0.20
+        assert penalty <= 0.35

tests/test_issue_detector.py

@@ -231,25 +231,25 @@ class TestDetectIssues:
     def test_task1_minimum_issues(self, dirty_conn, dirty_profile):
         records = copy.deepcopy(DIRTY_RECORDS)
         issues = detect_issues(dirty_conn, dirty_profile, records,
-                               task_id="task1_null_and_types", seed=42)
-        assert len(issues) >= MINIMUM_ISSUES["task1_null_and_types"]
+                               task_id="viz_easy", seed=42)
+        assert len(issues) >= MINIMUM_ISSUES["viz_easy"]
 
     def test_task2_minimum_issues(self, dirty_conn, dirty_profile):
         records = copy.deepcopy(DIRTY_RECORDS)
         issues = detect_issues(dirty_conn, dirty_profile, records,
-                               task_id="task2_constraints_and_fk", seed=42)
-        assert len(issues) >= MINIMUM_ISSUES["task2_constraints_and_fk"]
+                               task_id="ml_medium", seed=42)
+        assert len(issues) >= MINIMUM_ISSUES["ml_medium"]
 
     def test_task3_minimum_issues(self, dirty_conn, dirty_profile):
         records = copy.deepcopy(DIRTY_RECORDS)
         issues = detect_issues(dirty_conn, dirty_profile, records,
-                               task_id="task3_full_audit_with_trap", seed=42)
-        assert len(issues) >= MINIMUM_ISSUES["task3_full_audit_with_trap"]
+                               task_id="bq_hard", seed=42)
+        assert len(issues) >= MINIMUM_ISSUES["bq_hard"]
 
     def test_task1_only_null_and_type_issues(self, dirty_conn, dirty_profile):
         records = copy.deepcopy(DIRTY_RECORDS)
         issues = detect_issues(dirty_conn, dirty_profile, records,
-                               task_id="task1_null_and_types", seed=42)
+                               task_id="viz_easy", seed=42)
         for iss in issues:
             assert iss.issue_type in ("null", "type_error"), (
                 f"task1 should only detect null/type_error, got {iss.issue_type}"
@@ -258,14 +258,14 @@ class TestDetectIssues:
     def test_no_duplicate_issue_ids(self, dirty_conn, dirty_profile):
         records = copy.deepcopy(DIRTY_RECORDS)
         issues = detect_issues(dirty_conn, dirty_profile, records,
-                               task_id="task3_full_audit_with_trap", seed=42)
+                               task_id="bq_hard", seed=42)
         ids = [i.issue_id for i in issues]
         assert len(ids) == len(set(ids)), "Duplicate issue_ids found"
 
     def test_confidence_in_range(self, dirty_conn, dirty_profile):
         records = copy.deepcopy(DIRTY_RECORDS)
         issues = detect_issues(dirty_conn, dirty_profile, records,
-                               task_id="task3_full_audit_with_trap", seed=42)
+                               task_id="bq_hard", seed=42)
         for iss in issues:
             assert 0.0 <= iss.confidence <= 1.0, (
                 f"Issue {iss.issue_id} has out-of-range confidence {iss.confidence}"
@@ -275,8 +275,8 @@ class TestDetectIssues:
         """Clean data triggers synthetic top-up to meet minimum."""
         records = copy.deepcopy(CLEAN_RECORDS)
         issues = detect_issues(clean_conn, clean_profile, records,
-                               task_id="task1_null_and_types", seed=42)
-        assert len(issues) >= MINIMUM_ISSUES["task1_null_and_types"]
+                               task_id="viz_easy", seed=42)
+        assert len(issues) >= MINIMUM_ISSUES["viz_easy"]
 
     def test_reproducible_with_same_seed(self, dirty_conn, dirty_profile):
         conn2 = _make_conn(DIRTY_RECORDS)
@@ -284,9 +284,9 @@ class TestDetectIssues:
         r1 = copy.deepcopy(DIRTY_RECORDS)
         r2 = copy.deepcopy(DIRTY_RECORDS)
         issues1 = detect_issues(dirty_conn, dirty_profile, r1,
-                                task_id="task1_null_and_types", seed=99)
+                                task_id="viz_easy", seed=99)
         issues2 = detect_issues(conn2, profile2, r2,
-                                task_id="task1_null_and_types", seed=99)
+                                task_id="viz_easy", seed=99)
         assert len(issues1) == len(issues2)
         conn2.close()
 
@@ -299,7 +299,7 @@ class TestDetectTrap:
     def test_trap_planted_for_task3(self, dirty_conn, dirty_profile):
         records = copy.deepcopy(DIRTY_RECORDS)
         issues = detect_issues(dirty_conn, dirty_profile, records,
-                               task_id="task3_full_audit_with_trap", seed=42)
+                               task_id="bq_hard", seed=42)
         trap = detect_trap(dirty_conn, dirty_profile, records, issues, seed=42)
         assert trap is not None
         assert isinstance(trap, Trap)
@@ -307,7 +307,7 @@ class TestDetectTrap:
     def test_trap_not_in_issue_registry(self, dirty_conn, dirty_profile):
         records = copy.deepcopy(DIRTY_RECORDS)
         issues = detect_issues(dirty_conn, dirty_profile, records,
-                               task_id="task3_full_audit_with_trap", seed=42)
+                               task_id="bq_hard", seed=42)
         trap = detect_trap(dirty_conn, dirty_profile, records, issues, seed=42)
         if trap is None:
             pytest.skip("No numeric column available for trap")
@@ -317,7 +317,7 @@ class TestDetectTrap:
     def test_trap_value_is_2x_original(self, dirty_conn, dirty_profile):
         records = copy.deepcopy(DIRTY_RECORDS)
         issues = detect_issues(dirty_conn, dirty_profile, records,
-                               task_id="task3_full_audit_with_trap", seed=42)
+                               task_id="bq_hard", seed=42)
         trap = detect_trap(dirty_conn, dirty_profile, records, issues, seed=42)
         if trap is None:
             pytest.skip("No numeric column available for trap")
@@ -327,7 +327,7 @@ class TestDetectTrap:
     def test_trap_written_to_sqlite(self, dirty_conn, dirty_profile):
         records = copy.deepcopy(DIRTY_RECORDS)
         issues = detect_issues(dirty_conn, dirty_profile, records,
-                               task_id="task3_full_audit_with_trap", seed=42)
+                               task_id="bq_hard", seed=42)
         trap = detect_trap(dirty_conn, dirty_profile, records, issues, seed=42)
         if trap is None:
             pytest.skip("No numeric column available for trap")