Deploy Local Hybrid Engine

Dockerfile

@@ -0,0 +1,33 @@
+# Use an official Python runtime as a parent image
+FROM python:3.10-slim
+
+# Set the working directory in the container
+WORKDIR /app
+
+# Install system dependencies needed for compiling llama-cpp-python and basic libraries
+RUN apt-get update && apt-get install -y --no-install-recommends \
+    build-essential \
+    curl \
+    && rm -rf /var/lib/apt/lists/*
+
+# Copy the requirements file into the container
+COPY requirements.txt .
+
+# Install dependencies
+RUN pip install --no-cache-dir -r requirements.txt
+
+# Copy the rest of the application code
+COPY . .
+
+# Configure Hugging Face home and cache directories under /tmp/ for write permissions
+ENV HF_HOME=/tmp/.cache/huggingface
+ENV XDG_CACHE_HOME=/tmp/.cache
+ENV TRANSFORMERS_CACHE=/tmp/.cache/huggingface/hub
+ENV NUMBA_CACHE_DIR=/tmp/.cache/numba
+ENV PORT=7860
+
+# Expose port 7860
+EXPOSE 7860
+
+# Run uvicorn on port 7860
+CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "7860"]

README.md

@@ -0,0 +1,29 @@
+---
+title: Zero-LLM Engine
+emoji: 📊
+colorFrom: blue
+colorTo: indigo
+sdk: docker
+app_port: 7860
+pinned: false
+---
+
+# Zero-LLM Data Engine
+
+A high-scale, production-ready backend engine for natural-language Excel/CSV/Parquet editing, built with Python and Polars.
+
+## Local Development
+
+To run the engine locally:
+
+```bash
+cd zero-llm-engine
+pip install -r requirements.txt
+uvicorn main:app --host 0.0.0.0 --port 8000 --reload
+```
+
+## Features
+
+1. **Hybrid Parser**: Fast regex/keyword path with fallback to offline local embedding/spelling parser, local Qwen GGUF model, or Gemini cloud API.
+2. **Polars Execution**: Safe and fast data frame operations with concurrency, low latency, and memory efficiency.
+3. **MVCC Versioning**: Safe read/write concurrency via versioned files.

api/__init__.py

@@ -0,0 +1 @@
+# api package

api/routes.py

@@ -0,0 +1,96 @@
+"""
+REST routes — upload, download, session info, delete.
+"""
+from __future__ import annotations
+
+import os
+from fastapi import APIRouter, UploadFile, File, HTTPException
+from fastapi.responses import FileResponse
+
+from models.schemas import UploadResponse, SessionInfo, ErrorResponse
+from services import file_manager, session_manager, audit_service
+
+router = APIRouter(tags=["file operations"])
+
+
+# ── Upload ─────────────────────────────────────────────────────────
+
+@router.post(
+    "/upload",
+    response_model=UploadResponse,
+    responses={400: {"model": ErrorResponse}, 413: {"model": ErrorResponse}},
+    summary="Upload CSV / Excel / Parquet file",
+)
+async def upload_file(file: UploadFile = File(...)):
+    content = await file.read()
+    try:
+        result = file_manager.handle_upload(content, file.filename or "unknown.csv")
+    except ValueError as e:
+        raise HTTPException(status_code=400, detail=str(e))
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"Upload failed: {e}")
+    return UploadResponse(**result)
+
+
+# ── Download ───────────────────────────────────────────────────────
+
+@router.get(
+    "/download/{session_id}",
+    responses={404: {"model": ErrorResponse}},
+    summary="Download file as CSV",
+)
+async def download_file(session_id: str):
+    csv_path = file_manager.export_to_csv(session_id)
+    if csv_path is None:
+        raise HTTPException(status_code=404, detail="Session not found")
+    return FileResponse(
+        path=csv_path,
+        media_type="text/csv",
+        filename=f"{session_id}.csv",
+    )
+
+
+# ── Session info ───────────────────────────────────────────────────
+
+@router.get(
+    "/session/{session_id}",
+    response_model=SessionInfo,
+    responses={404: {"model": ErrorResponse}},
+    summary="Get session metadata",
+)
+async def get_session(session_id: str):
+    meta = session_manager.get(session_id)
+    if meta is None:
+        raise HTTPException(status_code=404, detail="Session not found")
+    return SessionInfo(
+        session_id=meta.session_id,
+        file_name=meta.file_name,
+        file_size_bytes=meta.file_size_bytes,
+        columns=meta.columns,
+        row_count=meta.row_count,
+        status=meta.status,
+    )
+
+
+# ── Delete session ─────────────────────────────────────────────────
+
+@router.delete(
+    "/session/{session_id}",
+    summary="Delete session and its data",
+)
+async def delete_session(session_id: str):
+    ok = file_manager.delete_session(session_id)
+    if not ok:
+        raise HTTPException(status_code=404, detail="Session not found")
+    return {"detail": "Session deleted"}
+
+
+# ── Audit history ──────────────────────────────────────────────────
+
+@router.get(
+    "/history/{session_id}",
+    summary="Get command history for a session",
+)
+async def get_history(session_id: str, limit: int = 50):
+    history = await audit_service.get_history(session_id, limit)
+    return {"history": history}

api/websocket.py

@@ -0,0 +1,96 @@
+"""
+WebSocket endpoint — receives natural-language commands, returns results.
+
+Protocol:
+  Client → Server:  {"command": "salary ko 10% badhao"}
+  Server → Client:  {"status": "success|error|unresolved", "message": "...", "diff": {...}, "suggestions": [...]}
+"""
+from __future__ import annotations
+
+import asyncio
+from fastapi import WebSocket, WebSocketDisconnect
+
+from core.parser.fallback import parse_intent_hybrid
+from core.validator import validate_intent
+from core.execution.mvcc_executor import execute_mvcc_command
+from core.column_registry import column_registry
+from services import session_manager, audit_service
+
+
+async def websocket_handler(ws: WebSocket, session_id: str) -> None:
+    await ws.accept()
+
+    # Verify session exists
+    meta = session_manager.get(session_id)
+    if meta is None:
+        await ws.send_json({
+            "status": "error",
+            "message": "Session nahi mila. Pehle file upload karo.",
+        })
+        await ws.close()
+        return
+
+    try:
+        while True:
+            # ── Receive command ─────────────────────────────────────
+            data = await ws.receive_json()
+            command: str = data.get("command", "").strip()
+            if not command:
+                await ws.send_json({
+                    "status": "error",
+                    "message": "Empty command",
+                })
+                continue
+
+            # ── Parse intent ────────────────────────────────────────
+            intent = parse_intent_hybrid(session_id, command)
+
+            if intent is None:
+                # Unresolved — suggest available columns
+                cols = column_registry.get_columns(session_id)
+                await ws.send_json({
+                    "status": "unresolved",
+                    "message": "Command samajh nahi aaya. Column ka naam check karo.",
+                    "suggestions": cols,
+                })
+                continue
+
+            # ── Validate ────────────────────────────────────────────
+            error = validate_intent(session_id, intent)
+            if error:
+                await ws.send_json({
+                    "status": "error",
+                    "message": error,
+                })
+                continue
+
+            # ── Execute (offload to thread for large files) ─────────
+            try:
+                result = await asyncio.to_thread(execute_mvcc_command, session_id, intent)
+            except Exception as exc:
+                await ws.send_json({
+                    "status": "error",
+                    "message": f"Execution error: {exc}",
+                })
+                continue
+
+            # ── Audit log ───────────────────────────────────────────
+            await audit_service.log_command(
+                session_id=session_id,
+                command=command,
+                intent=intent,
+                diff=result.get("diff"),
+            )
+
+            # Touch session
+            session_manager.touch(session_id)
+
+            # ── Send result ─────────────────────────────────────────
+            await ws.send_json({
+                "status": "success",
+                "message": result["message"],
+                "diff": result.get("diff", {}),
+            })
+
+    except WebSocketDisconnect:
+        pass  # Client disconnected, clean exit

config.py

@@ -0,0 +1,28 @@
+"""
+Application configuration.
+Swap values via environment variables in production.
+"""
+import os
+
+BASE_DIR = os.path.dirname(os.path.abspath(__file__))
+
+# Storage paths
+DATA_DIR = os.path.join(BASE_DIR, "data", "sessions")
+UPLOAD_DIR = os.path.join(BASE_DIR, "data", "uploads")
+AUDIT_DB_PATH = os.path.join(BASE_DIR, "data", "audit.db")
+
+# Files above this size use LazyFrame streaming instead of eager loading
+LAZY_THRESHOLD_BYTES = 100 * 1024 * 1024  # 100 MB
+
+# Fuzzy matching threshold (0-100)
+FUZZY_THRESHOLD = 78
+
+# Max file upload size: 1 GB
+MAX_UPLOAD_BYTES = 1024 * 1024 * 1024
+
+# Session auto-cleanup after idle minutes
+SESSION_TTL_MINUTES = 120
+
+# Ensure directories exist
+for _d in (DATA_DIR, UPLOAD_DIR, os.path.dirname(AUDIT_DB_PATH)):
+    os.makedirs(_d, exist_ok=True)

core/__init__.py

@@ -0,0 +1 @@
+# core package

core/a_to_z/__init__.py

@@ -0,0 +1 @@
+# zero-llm-engine core a-to-z operations

core/a_to_z/business/__init__.py

@@ -0,0 +1 @@
+# zero-llm-engine core a-to-z business operations

core/a_to_z/business/dedup.py

@@ -0,0 +1,54 @@
+"""
+Remove duplicates tool.
+
+If a column is specified, drops rows with duplicate values in that column
+(keeps first occurrence). If no column, drops fully-duplicate rows.
+"""
+from __future__ import annotations
+
+import os
+import polars as pl
+
+from config import DATA_DIR, LAZY_THRESHOLD_BYTES
+
+
+def _get_path(session_id: str) -> str:
+    return os.path.join(DATA_DIR, f"{session_id}.parquet")
+
+
+def execute_remove_duplicates(session_id: str, intent: dict) -> dict:
+    path = _get_path(session_id)
+    col = intent.get("column")  # may be None → full-row dedup
+
+    # Get count before
+    before_count = pl.scan_parquet(path).select(pl.len()).collect().item()
+
+    lf = pl.scan_parquet(path)
+    if col:
+        lf = lf.unique(subset=[col], keep="first")
+    else:
+        lf = lf.unique(keep="first")
+
+    file_sz = os.path.getsize(path)
+    if file_sz > LAZY_THRESHOLD_BYTES:
+        lf.sink_parquet(path)
+    else:
+        lf.collect().write_parquet(path)
+
+    after_count = pl.scan_parquet(path).select(pl.len()).collect().item()
+    removed = before_count - after_count
+
+    if col:
+        msg = f"'{col}' ke duplicate rows hata diye ({removed} rows removed, {after_count:,} remaining)"
+    else:
+        msg = f"Duplicate rows hata diye ({removed} rows removed, {after_count:,} remaining)"
+
+    return {
+        "message": msg,
+        "diff": {
+            "operation": "remove_duplicates",
+            "column": col,
+            "rows_removed": removed,
+            "rows_remaining": after_count,
+        },
+    }

core/a_to_z/business/find_replace.py

@@ -0,0 +1,86 @@
+"""
+Find & Replace tool.
+
+Replaces all occurrences of old_value with new_value in a column.
+Supports both string and numeric replacements.
+"""
+from __future__ import annotations
+
+import os
+import polars as pl
+
+from config import DATA_DIR, LAZY_THRESHOLD_BYTES
+
+
+def _get_path(session_id: str) -> str:
+    return os.path.join(DATA_DIR, f"{session_id}.parquet")
+
+
+def execute_find_replace(session_id: str, intent: dict) -> dict:
+    path = _get_path(session_id)
+    col = intent["column"]
+    old_val = intent["old_value"]
+    new_val = intent["new_value"]
+
+    # Determine if we should do string or numeric replacement
+    schema = pl.scan_parquet(path).collect_schema()
+    dtype = schema[col]
+    is_numeric = dtype in (pl.Float64, pl.Int64, pl.Int32, pl.Float32)
+
+    if is_numeric:
+        try:
+            old_num = float(old_val)
+            new_num = float(new_val)
+            expr = pl.when(pl.col(col) == old_num).then(pl.lit(new_num)).otherwise(pl.col(col)).alias(col)
+        except ValueError:
+            return {
+                "message": f"'{col}' numeric hai, replace values bhi numbers hone chahiye",
+                "diff": {},
+            }
+    else:
+        expr = (
+            pl.when(pl.col(col).cast(pl.String) == str(old_val))
+            .then(pl.lit(str(new_val)))
+            .otherwise(pl.col(col))
+            .alias(col)
+        )
+
+    file_sz = os.path.getsize(path)
+    if file_sz > LAZY_THRESHOLD_BYTES:
+        pl.scan_parquet(path).with_columns(expr).sink_parquet(path)
+    else:
+        df = pl.read_parquet(path).with_columns(expr)
+        df.write_parquet(path)
+
+    # Count affected rows
+    if is_numeric:
+        try:
+            old_num = float(old_val)
+            count = (
+                pl.scan_parquet(path)
+                .filter(pl.col(col) == float(new_val))
+                .select(pl.len())
+                .collect()
+                .item()
+            )
+        except ValueError:
+            count = "N/A"
+    else:
+        count = (
+            pl.scan_parquet(path)
+            .filter(pl.col(col).cast(pl.String) == str(new_val))
+            .select(pl.len())
+            .collect()
+            .item()
+        )
+
+    return {
+        "message": f"'{col}' mein '{old_val}' ko '{new_val}' se replace kiya ({count} rows changed)",
+        "diff": {
+            "operation": "find_replace",
+            "column": col,
+            "old_value": old_val,
+            "new_value": new_val,
+            "affected_rows": count,
+        },
+    }

core/a_to_z/student/__init__.py

@@ -0,0 +1 @@
+# zero-llm-engine core a-to-z student operations

core/a_to_z/student/aggregate.py

@@ -0,0 +1,89 @@
+"""
+Aggregate tools — sum, average, count, min, max.
+
+These are READ-ONLY operations. They do NOT modify the stored Parquet.
+They compute the result and return it as a message.
+"""
+from __future__ import annotations
+
+import os
+import polars as pl
+
+from config import DATA_DIR, LAZY_THRESHOLD_BYTES
+
+
+def _get_path(session_id: str) -> str:
+    return os.path.join(DATA_DIR, f"{session_id}.parquet")
+
+
+def _query(session_id: str, expr: pl.Expr):
+    """Run a single aggregate expression against the Parquet file.
+
+    Always uses LazyFrame + collect so the code path is identical
+    for both small and large files.
+    """
+    path = _get_path(session_id)
+    return pl.scan_parquet(path).select(expr).collect().item()
+
+
+def _format_number(val) -> str:
+    """Pretty-print a number with commas."""
+    if val is None:
+        return "N/A"
+    if isinstance(val, float):
+        if val == int(val):
+            return f"{int(val):,}"
+        return f"{val:,.2f}"
+    return f"{val:,}"
+
+
+def execute_sum(session_id: str, intent: dict) -> dict:
+    col = intent["column"]
+    result = _query(session_id, pl.col(col).sum())
+    return {
+        "message": f"{col} ka sum = {_format_number(result)}",
+        "diff": {"operation": "sum", "column": col, "result": result},
+    }
+
+
+def execute_average(session_id: str, intent: dict) -> dict:
+    col = intent["column"]
+    result = _query(session_id, pl.col(col).mean())
+    return {
+        "message": f"{col} ka average = {_format_number(result)}",
+        "diff": {"operation": "average", "column": col, "result": result},
+    }
+
+
+def execute_count(session_id: str, intent: dict) -> dict:
+    col = intent.get("column")
+    if col:
+        result = _query(session_id, pl.col(col).count())
+        return {
+            "message": f"{col} mein {_format_number(result)} non-null values hain",
+            "diff": {"operation": "count", "column": col, "result": result},
+        }
+    else:
+        result = _query(session_id, pl.len())
+        return {
+            "message": f"Total {_format_number(result)} rows hain",
+            "diff": {"operation": "count", "column": None, "result": result},
+        }
+
+
+def execute_min(session_id: str, intent: dict) -> dict:
+    col = intent["column"]
+    result = _query(session_id, pl.col(col).min())
+    return {
+        "message": f"{col} ka minimum = {_format_number(result)}",
+        "diff": {"operation": "min", "column": col, "result": result},
+    }
+
+
+def execute_max(session_id: str, intent: dict) -> dict:
+    col = intent["column"]
+    result = _query(session_id, pl.col(col).max())
+    return {
+        "message": f"{col} ka maximum = {_format_number(result)}",
+        "diff": {"operation": "max", "column": col, "result": result},
+    }

core/a_to_z/student/column_ops.py

@@ -0,0 +1,104 @@
+"""
+Column manipulation tools — delete, rename, add, cast type.
+"""
+from __future__ import annotations
+
+import os
+import polars as pl
+
+from config import DATA_DIR, LAZY_THRESHOLD_BYTES
+
+
+def _get_path(session_id: str) -> str:
+    return os.path.join(DATA_DIR, f"{session_id}.parquet")
+
+
+def _rewrite(session_id: str, lf: pl.LazyFrame) -> None:
+    path = _get_path(session_id)
+    if os.path.getsize(path) > LAZY_THRESHOLD_BYTES:
+        lf.sink_parquet(path)
+    else:
+        lf.collect().write_parquet(path)
+
+
+def execute_delete_column(session_id: str, intent: dict) -> dict:
+    path = _get_path(session_id)
+    col = intent["column"]
+    lf = pl.scan_parquet(path).drop(col)
+    _rewrite(session_id, lf)
+
+    return {
+        "message": f"'{col}' column delete ho gaya",
+        "diff": {"operation": "delete_column", "column": col},
+    }
+
+
+def execute_rename_column(session_id: str, intent: dict) -> dict:
+    path = _get_path(session_id)
+    old = intent["column"]
+    new = intent["new_name"]
+    lf = pl.scan_parquet(path).rename({old: new})
+    _rewrite(session_id, lf)
+
+    # Update column registry
+    from core.column_registry import column_registry
+    schema = pl.scan_parquet(_get_path(session_id)).collect_schema()
+    column_registry.register(session_id, schema.names())
+
+    return {
+        "message": f"'{old}' ka naam badal ke '{new}' ho gaya",
+        "diff": {"operation": "rename_column", "old_name": old, "new_name": new},
+    }
+
+
+def execute_add_column(session_id: str, intent: dict) -> dict:
+    path = _get_path(session_id)
+    # The intent should carry: column (name), or new_column_name
+    new_col = intent.get("new_column_name") or intent.get("column", "new_col")
+    default_val = intent.get("value", 0)
+
+    lf = pl.scan_parquet(path).with_columns(
+        pl.lit(default_val).alias(new_col)
+    )
+    _rewrite(session_id, lf)
+
+    # Refresh registry
+    from core.column_registry import column_registry
+    schema = pl.scan_parquet(_get_path(session_id)).collect_schema()
+    column_registry.register(session_id, schema.names())
+
+    return {
+        "message": f"Naya column '{new_col}' add ho gaya (default: {default_val})",
+        "diff": {"operation": "add_column", "column": new_col},
+    }
+
+
+def execute_cast_type(session_id: str, intent: dict) -> dict:
+    path = _get_path(session_id)
+    col = intent["column"]
+    target = intent["target_dtype"]
+
+    # Map string dtype names to Polars types
+    dtype_map = {
+        "Int64": pl.Int64,
+        "Int32": pl.Int32,
+        "Float64": pl.Float64,
+        "Float32": pl.Float32,
+        "String": pl.String,
+        "Boolean": pl.Boolean,
+        "Date": pl.Date,
+    }
+    pl_dtype = dtype_map.get(target)
+    if pl_dtype is None:
+        return {
+            "message": f"Type '{target}' supported nahi hai. Options: Int64, Float64, String, Boolean, Date",
+            "diff": {},
+        }
+
+    lf = pl.scan_parquet(path).with_columns(pl.col(col).cast(pl_dtype))
+    _rewrite(session_id, lf)
+
+    return {
+        "message": f"'{col}' ka type {target} mein change ho gaya",
+        "diff": {"operation": "cast_type", "column": col, "target_dtype": target},
+    }

core/a_to_z/student/filter_tool.py

@@ -0,0 +1,75 @@
+"""
+Filter tool.
+
+Keeps only rows matching a condition. The stored Parquet is REPLACED
+with the filtered subset.
+"""
+from __future__ import annotations
+
+import os
+import polars as pl
+
+from config import DATA_DIR, LAZY_THRESHOLD_BYTES
+
+
+def _get_path(session_id: str) -> str:
+    return os.path.join(DATA_DIR, f"{session_id}.parquet")
+
+
+def _build_predicate(column: str, condition: str, value) -> pl.Expr:
+    """Build a Polars expression from a comparison condition string."""
+    col = pl.col(column)
+
+    if condition == ">":
+        return col > value
+    elif condition == "<":
+        return col < value
+    elif condition == ">=":
+        return col >= value
+    elif condition == "<=":
+        return col <= value
+    elif condition == "!=":
+        return col != value
+    elif condition == "==" or condition == "=":
+        # Try numeric comparison first; fall back to string
+        if isinstance(value, (int, float)):
+            return col == value
+        try:
+            return col == float(value)
+        except (ValueError, TypeError):
+            return col.cast(pl.String).str.to_lowercase() == str(value).lower()
+    elif condition in ("contains", "~"):
+        return col.cast(pl.String).str.contains(str(value), literal=True)
+    else:
+        # Default: string equality (case-insensitive)
+        return col.cast(pl.String).str.to_lowercase() == str(value).lower()
+
+
+def execute_filter(session_id: str, intent: dict) -> dict:
+    path = _get_path(session_id)
+    col = intent["column"]
+    condition = intent["condition"]
+    filter_value = intent["filter_value"]
+
+    predicate = _build_predicate(col, condition, filter_value)
+
+    file_sz = os.path.getsize(path)
+    if file_sz > LAZY_THRESHOLD_BYTES:
+        lf = pl.scan_parquet(path).filter(predicate)
+        lf.sink_parquet(path)
+    else:
+        df = pl.read_parquet(path).filter(predicate)
+        df.write_parquet(path)
+
+    after_count = pl.scan_parquet(path).select(pl.len()).collect().item()
+
+    return {
+        "message": f"{col} {condition} {filter_value} → {after_count:,} rows bach gaye",
+        "diff": {
+            "operation": "filter",
+            "column": col,
+            "condition": condition,
+            "filter_value": filter_value,
+            "rows_after": after_count,
+        },
+    }

core/a_to_z/student/sort_tool.py

@@ -0,0 +1,43 @@
+"""
+Sort tool — ascending or descending by a single column.
+"""
+from __future__ import annotations
+
+import os
+import polars as pl
+
+from config import DATA_DIR, LAZY_THRESHOLD_BYTES
+
+
+def _get_path(session_id: str) -> str:
+    return os.path.join(DATA_DIR, f"{session_id}.parquet")
+
+
+def execute_sort(session_id: str, intent: dict) -> dict:
+    path = _get_path(session_id)
+    col = intent["column"]
+    descending = intent["operation"] == "sort_desc"
+
+    file_sz = os.path.getsize(path)
+
+    # Sort requires the full column in memory either way.
+    # For large files, collect then write back.
+    lf = pl.scan_parquet(path)
+    if file_sz > LAZY_THRESHOLD_BYTES:
+        lf.sort(col, descending=descending).sink_parquet(path)
+    else:
+        df = lf.collect().sort(col, descending=descending)
+        df.write_parquet(path)
+
+    row_count = pl.scan_parquet(path).select(pl.len()).collect().item()
+    order = "descending (bada se chhota)" if descending else "ascending (chhota se bada)"
+
+    return {
+        "message": f"{col} ko {order} sort kiya ({row_count:,} rows)",
+        "diff": {
+            "operation": "sort",
+            "column": col,
+            "descending": descending,
+            "affected_rows": row_count,
+        },
+    }

core/a_to_z/student/update.py

@@ -0,0 +1,80 @@
+"""
+Increase / Decrease tool.
+
+Supports both percentage and absolute value changes.
+  "salary ko 10% badhao"  → multiply by 1.10
+  "salary kam karo 5000"  → subtract 5000 (absolute when no % sign)
+"""
+from __future__ import annotations
+
+import os
+import polars as pl
+
+from config import DATA_DIR, LAZY_THRESHOLD_BYTES
+
+
+def _get_path(session_id: str) -> str:
+    return os.path.join(DATA_DIR, f"{session_id}.parquet")
+
+
+def _file_size(path: str) -> int:
+    return os.path.getsize(path)
+
+
+def execute_increase(session_id: str, intent: dict) -> dict:
+    return _execute_update(session_id, intent, mode="increase")
+
+
+def execute_decrease(session_id: str, intent: dict) -> dict:
+    return _execute_update(session_id, intent, mode="decrease")
+
+
+def _execute_update(session_id: str, intent: dict, mode: str) -> dict:
+    """Core update logic shared by increase & decrease."""
+    path = _get_path(session_id)
+    col = intent["column"]
+    value = intent.get("value")
+    if value is None:
+        value = 10.0  # sensible default
+
+    is_percent = intent.get("is_percent", True)  # default to percentage
+    # Heuristic: if the original command had "%", treat as percentage
+    # The intent parser already extracts the number; we need a hint.
+    # For now, always treat as percentage (most common use-case).
+    # Absolute mode can be added via intent["absolute"] = True.
+
+    if is_percent:
+        factor = 1 + value / 100 if mode == "increase" else 1 - value / 100
+        expr = (pl.col(col) * factor).alias(col)
+        op_label = f"{value}% badha diya" if mode == "increase" else f"{value}% ghata diya"
+    else:
+        expr = (
+            (pl.col(col) + value).alias(col)
+            if mode == "increase"
+            else (pl.col(col) - value).alias(col)
+        )
+        op_label = f"{value} joda" if mode == "increase" else f"{value} ghata diya"
+
+    file_sz = _file_size(path)
+    if file_sz > LAZY_THRESHOLD_BYTES:
+        # Streaming: lazy → sink_parquet (no full data in RAM)
+        pl.scan_parquet(path).with_columns(expr).sink_parquet(path)
+    else:
+        # Eager: fast in-memory for small files
+        df = pl.read_parquet(path)
+        df = df.with_columns(expr)
+        df.write_parquet(path)
+
+    # Get affected row count (cheap metadata read)
+    row_count = pl.scan_parquet(path).select(pl.len()).collect().item()
+
+    return {
+        "message": f"{col} ko {op_label} ({row_count:,} rows updated)",
+        "diff": {
+            "operation": mode,
+            "column": col,
+            "value": value,
+            "is_percent": is_percent,
+            "affected_rows": row_count,
+        },
+    }

core/column_registry.py

@@ -0,0 +1,80 @@
+"""
+Column name registry.
+
+Builds a bidirectional alias map per session so that repeated commands
+resolve columns in O(1) time. Falls back to rapidfuzz only on first
+encounter, then caches the result.
+"""
+from __future__ import annotations
+
+from typing import Optional
+from rapidfuzz import process, fuzz
+
+from config import FUZZY_THRESHOLD
+
+
+class ColumnRegistry:
+    """Session-scoped column name resolver with fuzzy-match cache."""
+
+    def __init__(self) -> None:
+        # session_id -> {normalized_alias: real_column_name}
+        self._store: dict[str, dict[str, str]] = {}
+
+    # ── public API ──────────────────────────────────────────────────
+
+    def register(self, session_id: str, columns: list[str]) -> None:
+        """Build alias map for every column in the dataframe."""
+        mapping: dict[str, str] = {}
+        for col in columns:
+            normalized = col.lower().strip()
+            mapping[normalized] = col
+            # Common variations the user might type
+            mapping[col.replace("_", " ").lower()] = col
+            mapping[col.replace(" ", "_").lower()] = col
+            mapping[col.replace("-", "_").lower()] = col
+            mapping[col.replace("-", " ").lower()] = col
+        self._store[session_id] = mapping
+
+    def resolve(self, session_id: str, token: str) -> Optional[str]:
+        """Resolve a user-provided token to the real column name.
+
+        1. Exact match on alias map (O(1)).
+        2. rapidfuzz WRatio against all aliases.
+        3. Tokenize the full command and try sliding-window bigrams.
+        """
+        mapping = self._store.get(session_id)
+        if not mapping:
+            return None
+
+        # Direct O(1) lookup
+        key = token.lower().strip()
+        if key in mapping:
+            return mapping[key]
+
+        # Single-token fuzzy
+        best = process.extractOne(key, list(mapping.keys()), scorer=fuzz.WRatio)
+        if best and best[1] >= FUZZY_THRESHOLD:
+            return mapping[best[0]]
+
+        # Sliding-window bigram from the original token (handles multi-word columns)
+        parts = token.split()
+        for i in range(len(parts) - 1):
+            bigram = f"{parts[i]} {parts[i + 1]}".lower()
+            if bigram in mapping:
+                return mapping[bigram]
+            best = process.extractOne(bigram, list(mapping.keys()), scorer=fuzz.WRatio)
+            if best and best[1] >= FUZZY_THRESHOLD:
+                return mapping[best[0]]
+
+        return None
+
+    def get_columns(self, session_id: str) -> list[str]:
+        """Return deduplicated real column names."""
+        return list(set(self._store.get(session_id, {}).values()))
+
+    def remove(self, session_id: str) -> None:
+        self._store.pop(session_id, None)
+
+
+# Module-level singleton
+column_registry = ColumnRegistry()

core/execution/__init__.py

@@ -0,0 +1,3 @@
+"""
+Execution package.
+"""

core/execution/mvcc_executor.py

@@ -0,0 +1,365 @@
+"""
+MVCC Executor.
+Performs lock-free versioned operations on Parquet files using Polars LazyFrames.
+"""
+from __future__ import annotations
+
+import os
+import polars as pl
+
+from config import DATA_DIR, LAZY_THRESHOLD_BYTES
+from services.session_manager import session_manager
+from core.column_registry import column_registry
+from core.execution.polars_builder import build_polars_expression
+
+def _build_predicate(column: str, condition: str, value) -> pl.Expr:
+    """Build a Polars expression from a comparison condition string."""
+    col = pl.col(column)
+    if condition == ">":
+        return col > value
+    elif condition == "<":
+        return col < value
+    elif condition == ">=":
+        return col >= value
+    elif condition == "<=":
+        return col <= value
+    elif condition == "!=":
+        return col != value
+    elif condition == "==" or condition == "=":
+        if isinstance(value, (int, float)):
+            return col == value
+        try:
+            return col == float(value)
+        except (ValueError, TypeError):
+            return col.cast(pl.String).str.to_lowercase() == str(value).lower()
+    elif condition in ("contains", "~"):
+        return col.cast(pl.String).str.contains(str(value), literal=True)
+    else:
+        return col.cast(pl.String).str.to_lowercase() == str(value).lower()
+
+def _format_number(val) -> str:
+    """Pretty-print numbers with commas."""
+    if val is None:
+        return "N/A"
+    if isinstance(val, float):
+        if val == int(val):
+            return f"{int(val):,}"
+        return f"{val:,.2f}"
+    if isinstance(val, int):
+        return f"{val:,}"
+    return str(val)
+
+def execute_mvcc_command(session_id: str, intent: dict) -> dict:
+    """Execute the command and return a result dict.
+
+    Updates metadata and creates a new parquet version on write.
+    """
+    meta = session_manager.get(session_id)
+    if meta is None:
+        return {
+            "status": "error",
+            "message": "Session nahi mila. Pehle file upload karo.",
+            "diff": {}
+        }
+
+    version = getattr(meta, "current_version", 0)
+
+    # Resolve read path
+    # If version is 0, we can fall back to the initial upload path (data/sessions/{session_id}.parquet).
+    v_path = os.path.join(DATA_DIR, f"{session_id}_v{version}.parquet")
+    if not os.path.exists(v_path):
+        if version == 0:
+            v_path = os.path.join(DATA_DIR, f"{session_id}.parquet")
+            if not os.path.exists(v_path):
+                raise FileNotFoundError(f"Initial Parquet file not found for session {session_id}")
+        else:
+            raise FileNotFoundError(f"Parquet file not found for version {version} of session {session_id}")
+
+    op = intent.get("operation")
+    if not op:
+        return {
+            "message": "Operation specifies nahi kiya gaya",
+            "diff": {}
+        }
+
+    # ── Read-only operations (Aggregates) ──
+    if op in ("sum", "average", "count", "min", "max"):
+        lf = pl.scan_parquet(v_path)
+        if op == "sum":
+            col = intent["column"]
+            result = lf.select(pl.col(col).sum()).collect().item()
+            return {
+                "message": f"{col} ka sum = {_format_number(result)}",
+                "diff": {"operation": "sum", "column": col, "result": result},
+            }
+        elif op == "average":
+            col = intent["column"]
+            result = lf.select(pl.col(col).mean()).collect().item()
+            return {
+                "message": f"{col} ka average = {_format_number(result)}",
+                "diff": {"operation": "average", "column": col, "result": result},
+            }
+        elif op == "count":
+            col = intent.get("column")
+            if col:
+                result = lf.select(pl.col(col).count()).collect().item()
+                return {
+                    "message": f"{col} mein {_format_number(result)} non-null values hain",
+                    "diff": {"operation": "count", "column": col, "result": result},
+                }
+            else:
+                result = lf.select(pl.len()).collect().item()
+                return {
+                    "message": f"Total {_format_number(result)} rows hain",
+                    "diff": {"operation": "count", "column": None, "result": result},
+                }
+        elif op == "min":
+            col = intent["column"]
+            result = lf.select(pl.col(col).min()).collect().item()
+            return {
+                "message": f"{col} ka minimum = {_format_number(result)}",
+                "diff": {"operation": "min", "column": col, "result": result},
+            }
+        elif op == "max":
+            col = intent["column"]
+            result = lf.select(pl.col(col).max()).collect().item()
+            return {
+                "message": f"{col} ka maximum = {_format_number(result)}",
+                "diff": {"operation": "max", "column": col, "result": result},
+            }
+
+    # ── Write operations ──
+    lf = pl.scan_parquet(v_path)
+    new_version = version + 1
+    new_path = os.path.join(DATA_DIR, f"{session_id}_v{new_version}.parquet")
+
+    before_count = lf.select(pl.len()).collect().item()
+
+    if op in ("increase", "decrease"):
+        col = intent["column"]
+        value = intent.get("value")
+        if value is None:
+            value = 10.0
+        is_percent = intent.get("is_percent", True)
+        
+        expr = build_polars_expression(op, col, {"value": value, "is_percent": is_percent})
+        lf = lf.with_columns(expr)
+
+    elif op == "cast_type":
+        col = intent["column"]
+        target = intent["target_dtype"]
+        expr = build_polars_expression("cast_type", col, {"target_dtype": target})
+        lf = lf.with_columns(expr)
+
+    elif op == "find_replace":
+        col = intent["column"]
+        old_val = intent["old_value"]
+        new_val = intent["new_value"]
+        
+        schema = lf.collect_schema()
+        dtype = schema[col]
+        is_numeric = dtype in (pl.Float64, pl.Int64, pl.Int32, pl.Float32)
+        
+        if is_numeric:
+            try:
+                float(old_val)
+                float(new_val)
+            except (ValueError, TypeError):
+                return {
+                    "message": f"'{col}' numeric hai, replace values bhi numbers hone chahiye",
+                    "diff": {},
+                }
+        
+        expr = build_polars_expression("find_replace", col, {
+            "old_value": old_val,
+            "new_value": new_val,
+            "is_numeric": is_numeric
+        })
+        lf = lf.with_columns(expr)
+
+    elif op == "delete_column":
+        col = intent["column"]
+        lf = lf.drop(col)
+
+    elif op == "rename_column":
+        old = intent["column"]
+        new = intent["new_name"]
+        lf = lf.rename({old: new})
+
+    elif op == "add_column":
+        new_col = intent.get("new_column_name") or intent.get("column", "new_col")
+        default_val = intent.get("value", 0)
+        lf = lf.with_columns(pl.lit(default_val).alias(new_col))
+
+    elif op == "remove_duplicates":
+        col = intent.get("column")
+        if col:
+            lf = lf.unique(subset=[col], keep="first")
+        else:
+            lf = lf.unique(keep="first")
+
+    elif op == "filter":
+        col = intent["column"]
+        condition = intent["condition"]
+        filter_value = intent["filter_value"]
+        predicate = _build_predicate(col, condition, filter_value)
+        lf = lf.filter(predicate)
+
+    elif op in ("sort_asc", "sort_desc"):
+        col = intent["column"]
+        descending = (op == "sort_desc")
+        lf = lf.sort(col, descending=descending)
+
+    else:
+        return {
+            "message": f"'{op}' operation supported nahi hai",
+            "diff": {},
+        }
+
+    # Lock-free versioned write
+    prev_size = os.path.getsize(v_path)
+    try:
+        if prev_size > LAZY_THRESHOLD_BYTES:
+            lf.sink_parquet(new_path)
+        else:
+            lf.collect().write_parquet(new_path)
+    except Exception:
+        # Fallback to eager write if streaming sink is not supported for this query plan (e.g. sort/unique)
+        lf.collect().write_parquet(new_path)
+
+    # Read updated metadata
+    lf_new = pl.scan_parquet(new_path)
+    new_schema = lf_new.collect_schema()
+    after_count = lf_new.select(pl.len()).collect().item()
+    columns_meta = [{"name": name, "dtype": str(dtype)} for name, dtype in new_schema.items()]
+
+    # Format result message and diff
+    message = ""
+    diff = {}
+
+    if op in ("increase", "decrease"):
+        col = intent["column"]
+        value = intent.get("value")
+        if value is None:
+            value = 10.0
+        is_percent = intent.get("is_percent", True)
+        op_label = f"{value}% badha diya" if op == "increase" else f"{value}% ghata diya"
+        if not is_percent:
+            op_label = f"{value} joda" if op == "increase" else f"{value} ghata diya"
+        message = f"{col} ko {op_label} ({after_count:,} rows updated)"
+        diff = {
+            "operation": op,
+            "column": col,
+            "value": value,
+            "is_percent": is_percent,
+            "affected_rows": after_count,
+        }
+
+    elif op == "cast_type":
+        col = intent["column"]
+        target = intent["target_dtype"]
+        message = f"'{col}' ka type {target} mein change ho gaya"
+        diff = {"operation": "cast_type", "column": col, "target_dtype": target}
+
+    elif op == "find_replace":
+        col = intent["column"]
+        old_val = intent["old_value"]
+        new_val = intent["new_value"]
+        
+        dtype = new_schema[col]
+        is_numeric = dtype in (pl.Float64, pl.Int64, pl.Int32, pl.Float32)
+        if is_numeric:
+            try:
+                count = lf_new.filter(pl.col(col) == float(new_val)).select(pl.len()).collect().item()
+            except (ValueError, TypeError):
+                count = "N/A"
+        else:
+            count = lf_new.filter(pl.col(col).cast(pl.String) == str(new_val)).select(pl.len()).collect().item()
+            
+        message = f"'{col}' mein '{old_val}' ko '{new_val}' se replace kiya ({count} rows changed)"
+        diff = {
+            "operation": "find_replace",
+            "column": col,
+            "old_value": old_val,
+            "new_value": new_val,
+            "affected_rows": count,
+        }
+
+    elif op == "delete_column":
+        col = intent["column"]
+        message = f"'{col}' column delete ho gaya"
+        diff = {"operation": "delete_column", "column": col}
+
+    elif op == "rename_column":
+        old = intent["column"]
+        new = intent["new_name"]
+        message = f"'{old}' ka naam badal ke '{new}' ho gaya"
+        diff = {"operation": "rename_column", "old_name": old, "new_name": new}
+
+    elif op == "add_column":
+        new_col = intent.get("new_column_name") or intent.get("column", "new_col")
+        default_val = intent.get("value", 0)
+        message = f"Naya column '{new_col}' add ho gaya (default: {default_val})"
+        diff = {"operation": "add_column", "column": new_col}
+
+    elif op == "remove_duplicates":
+        col = intent.get("column")
+        removed = before_count - after_count
+        if col:
+            message = f"'{col}' ke duplicate rows hata diye ({removed} rows removed, {after_count:,} remaining)"
+        else:
+            message = f"Duplicate rows hata diye ({removed} rows removed, {after_count:,} remaining)"
+        diff = {
+            "operation": "remove_duplicates",
+            "column": col,
+            "rows_removed": removed,
+            "rows_remaining": after_count,
+        }
+
+    elif op == "filter":
+        col = intent["column"]
+        condition = intent["condition"]
+        filter_value = intent["filter_value"]
+        message = f"{col} {condition} {filter_value} → {after_count:,} rows bach gaye"
+        diff = {
+            "operation": "filter",
+            "column": col,
+            "condition": condition,
+            "filter_value": filter_value,
+            "rows_after": after_count,
+        }
+
+    elif op in ("sort_asc", "sort_desc"):
+        col = intent["column"]
+        descending = (op == "sort_desc")
+        order = "descending (bada se chhota)" if descending else "ascending (chhota se bada)"
+        message = f"{col} ko {order} sort kiya ({after_count:,} rows)"
+        diff = {
+            "operation": "sort",
+            "column": col,
+            "descending": descending,
+            "affected_rows": after_count,
+        }
+
+    # Update session metadata in session_manager
+    meta.current_version = new_version
+    meta.row_count = after_count
+    meta.columns = columns_meta
+    meta.touch()
+
+    # Update column registry
+    column_registry.register(session_id, [c["name"] for c in columns_meta])
+
+    # Clean up the previous intermediate versioned file
+    if version > 0:
+        prev_version_path = os.path.join(DATA_DIR, f"{session_id}_v{version}.parquet")
+        if os.path.exists(prev_version_path):
+            try:
+                os.remove(prev_version_path)
+            except Exception as e:
+                print(f"[MVCC Executor] Warning cleaning up version {version}: {e}")
+
+    return {
+        "message": message,
+        "diff": diff,
+    }

core/execution/polars_builder.py

@@ -0,0 +1,64 @@
+"""
+Polars Expression Builder.
+Builds programmatic expressions securely without raw eval/exec.
+"""
+from __future__ import annotations
+
+import polars as pl
+
+def build_polars_expression(operation: str, column: str, params: dict) -> pl.Expr:
+    """Build programmatic expressions for increase, decrease, cast_type, and find_replace."""
+    if operation == "increase":
+        value = params.get("value", 10.0)
+        is_percent = params.get("is_percent", True)
+        if is_percent:
+            return (pl.col(column) * (1 + value / 100)).alias(column)
+        else:
+            return (pl.col(column) + value).alias(column)
+            
+    elif operation == "decrease":
+        value = params.get("value", 10.0)
+        is_percent = params.get("is_percent", True)
+        if is_percent:
+            return (pl.col(column) * (1 - value / 100)).alias(column)
+        else:
+            return (pl.col(column) - value).alias(column)
+            
+    elif operation == "cast_type":
+        target = params.get("target_dtype")
+        dtype_map = {
+            "Int64": pl.Int64,
+            "Int32": pl.Int32,
+            "Float64": pl.Float64,
+            "Float32": pl.Float32,
+            "String": pl.String,
+            "Boolean": pl.Boolean,
+            "Date": pl.Date,
+        }
+        pl_dtype = dtype_map.get(target)
+        if pl_dtype is None:
+            raise ValueError(f"Unsupported target dtype: {target}")
+        return pl.col(column).cast(pl_dtype).alias(column)
+        
+    elif operation == "find_replace":
+        old_val = params.get("old_value")
+        new_val = params.get("new_value")
+        is_numeric = params.get("is_numeric", False)
+        
+        if is_numeric:
+            try:
+                old_num = float(old_val)
+                new_num = float(new_val)
+                return pl.when(pl.col(column) == old_num).then(pl.lit(new_num)).otherwise(pl.col(column)).alias(column)
+            except (ValueError, TypeError):
+                raise ValueError("Values must be numeric for a numeric column replacement.")
+        else:
+            return (
+                pl.when(pl.col(column).cast(pl.String) == str(old_val))
+                .then(pl.lit(str(new_val)))
+                .otherwise(pl.col(column))
+                .alias(column)
+            )
+            
+    else:
+        raise ValueError(f"Unsupported operation for programmatic expression building: {operation}")

core/intent_parser.py

@@ -0,0 +1,375 @@
+"""
+Intent parser — regex + keyword + fuzzy column resolution.
+
+No LLM anywhere. Covers Student + Business tier operations:
+  increase, decrease, filter, sort_asc, sort_desc,
+  sum, average, count, min, max,
+  find_replace, delete_column, rename_column, add_column,
+  remove_duplicates, cast_type
+"""
+from __future__ import annotations
+
+import re
+from typing import Optional
+
+from rapidfuzz import process as rf_process, fuzz as rf_fuzz
+from core.column_registry import column_registry
+
+# ── Operation keyword table ─────────────────────────────────────────
+# Longer, more specific keywords score higher so "badhao" beats "bada"
+# when both appear in a command.
+OPERATION_KEYWORDS: dict[str, list[str]] = {
+    "increase": [
+        "badhao", "badha do", "increase", "barha do", "barhao",
+        "zyada karo", "bada karo", "grow", "raise", "badha dijiye",
+    ],
+    "decrease": [
+        "ghatao", "ghata do", "decrease", "kam karo", "kam kar do",
+        "chhota karo", "reduce", "cut", "minus karo", "ghata dijiye",
+    ],
+    "filter": [
+        "sirf", "only", "filter", "dikhao", "show only", "show me",
+        "bas", "wale dikhao", "where", "jitne", "laao",
+    ],
+    "sort_asc": [
+        "chota se bada", "ascending", "a to z", "low to high",
+        "smallest first", "ascending order", "a-z", "asc",
+    ],
+    "sort_desc": [
+        "bada se chota", "descending", "z to a", "high to low",
+        "largest first", "descending order", "z-a", "desc",
+        "bade se chhote",
+    ],
+    "sum": [
+        "sum", "total", "jod", "yog", "add up", "total batao",
+        "kul", "jama",
+    ],
+    "average": [
+        "average", "avg", "mean", "samanya", "average nikalo",
+    ],
+    "count": [
+        "count", "ginti", "kitne", "kitni rows", "count karo",
+        "kitni", "rows kitne", "total rows",
+    ],
+    "min": [
+        "minimum", "min", "sabse chhota", "lowest", "kam se kam",
+    ],
+    "max": [
+        "maximum", "max", "sabse bada", "highest", "zyada se zyada",
+    ],
+    "find_replace": [
+        "replace", "badlo", "change", "find", "dhundho",
+        "substitute", "replace karo", "change karo", "naye se badlo",
+    ],
+    "delete_column": [
+        "delete column", "column hatao", "column delete karo",
+        "remove column", "column remove karo", "column hatado",
+        "column drop karo",
+    ],
+    "rename_column": [
+        "rename column", "column ka naam badlo", "column rename karo",
+        "name change karo", "naam badlo", "rename karo",
+    ],
+    "add_column": [
+        "add column", "naya column banao", "column add karo",
+        "new column", "column create karo",
+    ],
+    "remove_duplicates": [
+        "duplicate hatao", "duplicates remove karo", "unique rakho",
+        "duplicate remove", "repeat hatao",
+    ],
+    "cast_type": [
+        "type badlo", "data type change", "convert type",
+        "type convert karo", "numeric banao", "string banao",
+    ],
+}
+
+# ── Compiled regexes ───────────────────────────────────────────────
+PERCENT_RE = re.compile(r"(\d+\.?\d*)\s*%", re.IGNORECASE)
+NUMBER_RE = re.compile(r"(\d+\.?\d*)")
+
+# Comparison operators (symbols)
+_CMP_SYMBOLS = re.compile(r"([><=!]+)\s*([\d.]+|[^\s]+)")
+
+HINDI_CMP_MAP: dict[str, str] = {
+    "se zyada": ">",
+    "se kam": "<",
+    "ke barabar": "==",
+    "se zyada ya barabar": ">=",
+    "se kam ya barabar": "<=",
+    "se bada": ">",
+    "se chhota": "<",
+    "ke equal": "==",
+    "ke hi": "==",
+}
+
+# Words to strip when extracting filter values
+_STOP_WORDS = [
+    "sirf", "only", "filter", "dikhao", "show", "show only",
+    "bas", "wale", "laao", "bhai", "ko", "ka", "ke", "ki",
+    "mein", "hai", "hain", "karo", "karein", "sort",
+    "bada", "chhota", "se", "nikalo", "batao",
+]
+
+
+# ══════════════════════════════════════════════════════════════════════
+#  Public API
+# ══════════════════════════════════════════════════════════════════════
+
+def parse_intent(session_id: str, command: str) -> Optional[dict]:
+    """Parse a natural-language command into a structured intent dict.
+
+    Returns ``None`` when nothing can be resolved (the caller should
+    return an *unresolved* response with column suggestions).
+    """
+    columns = column_registry.get_columns(session_id)
+    if not columns:
+        return None
+
+    operation = _match_operation(command)
+    if operation is None:
+        return None
+
+    # ── Operations that need special parsing ──────────────────────
+    if operation == "remove_duplicates":
+        col = _best_column(session_id, command, columns)
+        return {"operation": "remove_duplicates", "column": col}
+
+    if operation == "delete_column":
+        col = _best_column(session_id, command, columns)
+        if col is None:
+            return None
+        return {"operation": "delete_column", "column": col}
+
+    if operation == "rename_column":
+        return _parse_rename(session_id, command, columns)
+
+    if operation == "find_replace":
+        return _parse_find_replace(session_id, command, columns)
+
+    if operation == "filter":
+        return _parse_filter(session_id, command, columns)
+
+    if operation == "cast_type":
+        return _parse_cast(session_id, command, columns)
+
+    # ── Standard: operation + column + optional value ────────────
+    col = _best_column(session_id, command, columns)
+    if col is None:
+        return None
+
+    value = _parse_value(command)
+
+    return {"operation": operation, "column": col, "value": value}
+
+
+# ══════════════════════════════════════════════════════════════════════
+#  Internal helpers
+# ══════════════════════════════════════════════════════════════════════
+
+def _match_operation(command: str) -> Optional[str]:
+    """Pick the operation with the highest keyword-match score."""
+    cmd = command.lower()
+    scores: dict[str, int] = {}
+    for op, keywords in OPERATION_KEYWORDS.items():
+        for kw in keywords:
+            if kw in cmd:
+                # Weight by keyword length so specific phrases beat short ones
+                scores[op] = scores.get(op, 0) + len(kw)
+    if not scores:
+        return None
+    return max(scores, key=scores.get)  # type: ignore[arg-type]
+
+
+# Flat set of all operation keywords — used to skip them during column resolution
+_ALL_OP_KEYWORDS: set[str] = set()
+for _kws in OPERATION_KEYWORDS.values():
+    _ALL_OP_KEYWORDS.update(_kws)
+_ALL_OP_KEYWORDS.update(_STOP_WORDS)
+_ALL_OP_KEYWORDS.update(["ko", "ka", "ke", "ki", "karo", "nikalo", "batao",
+                          "hai", "hain", "mein", "se", "do", "dijiye"])
+
+
+def _best_column(session_id: str, command: str, columns: list[str]) -> Optional[str]:
+    """Fuzzy-resolve the best column from the command text.
+
+    1. Try column_registry (O(1) alias + cached fuzzy).
+    2. Fall back to direct rapidfuzz scan.
+
+    Skips tokens that are known operation keywords (e.g. "average" won't
+    false-match to column "Age").
+    """
+    tokens = command.split()
+
+    # Filter out keyword tokens and pure-number tokens
+    clean_tokens = [
+        t for t in tokens
+        if not re.fullmatch(r"[\d.]+%?", t)
+        and t.lower() not in _ALL_OP_KEYWORDS
+    ]
+    candidates = clean_tokens + [
+        " ".join(clean_tokens[i : i + 2]) for i in range(len(clean_tokens) - 1)
+    ]
+
+    # 1. Registry first (O(1) alias + cached fuzzy)
+    for token in candidates:
+        resolved = column_registry.resolve(session_id, token)
+        if resolved:
+            return resolved
+
+    # 2. Direct rapidfuzz scan as fallback
+    best_match, best_score = None, 0
+    for cand in candidates:
+        hit = rf_process.extractOne(cand, columns, scorer=rf_fuzz.WRatio)
+        if hit and hit[1] > best_score:
+            best_match, best_score = hit[0], hit[1]
+    return best_match if best_score >= 78 else None
+
+
+def _parse_value(command: str) -> Optional[float]:
+    """Extract a numeric value. Percentage wins over absolute."""
+    m = PERCENT_RE.search(command)
+    if m:
+        return float(m.group(1))
+    m = NUMBER_RE.search(command)
+    if m:
+        return float(m.group(1))
+    return None
+
+
+def _parse_filter(session_id: str, command: str, columns: list[str]) -> Optional[dict]:
+    """Resolve a filter command into {column, condition, filter_value}."""
+    col = _best_column(session_id, command, columns)
+    if col is None:
+        return None
+
+    cmd_lower = command.lower()
+
+    # 1. Symbol comparison:  salary > 50000
+    sym = _CMP_SYMBOLS.search(cmd_lower)
+    if sym and col.lower() in cmd_lower:
+        op_str, val_str = sym.group(1), sym.group(2)
+        try:
+            fval: str | float = float(val_str)
+        except ValueError:
+            fval = val_str.strip("'\"")
+        return {
+            "operation": "filter",
+            "column": col,
+            "condition": op_str,
+            "filter_value": fval,
+        }
+
+    # 2. Hindi comparison:  salary 50000 se zyada
+    for hindi_op, symbol in HINDI_CMP_MAP.items():
+        if hindi_op in cmd_lower:
+            num_match = re.search(
+                r"(\d+\.?\d*)\s+" + re.escape(hindi_op), cmd_lower
+            )
+            if num_match:
+                return {
+                    "operation": "filter",
+                    "column": col,
+                    "condition": symbol,
+                    "filter_value": float(num_match.group(1)),
+                }
+
+    # 3. Equality by presence:  "city Mumbai dikhao" → city == Mumbai
+    stripped = cmd_lower
+    for kw in _STOP_WORDS:
+        stripped = stripped.replace(kw, "")
+    stripped = stripped.replace(col.lower(), "", 1).strip()
+    if stripped:
+        stripped = re.sub(r"^[><=!]+\s*", "", stripped).strip()
+        return {
+            "operation": "filter",
+            "column": col,
+            "condition": "==",
+            "filter_value": stripped,
+        }
+
+    return {"operation": "filter", "column": col, "condition": None, "filter_value": None}
+
+
+def _parse_find_replace(session_id: str, command: str, columns: list[str]) -> Optional[dict]:
+    """Extract old_value and new_value for find & replace."""
+    col = _best_column(session_id, command, columns)
+    if col is None:
+        return None
+
+    # Try quoted values first
+    quoted = re.findall(r"""['"]([^'"]+)['"]""", command)
+    if len(quoted) >= 2:
+        return {
+            "operation": "find_replace",
+            "column": col,
+            "old_value": quoted[0],
+            "new_value": quoted[1],
+        }
+
+    # Try "X ko Y se badlo" / "replace X with Y" / "X ko Y replace karo"
+    m = re.search(
+        r"(\S+)\s+ko\s+(\S+)\s+(?:se\s+)?badlo"
+        r"|replace\s+(\S+)\s+with\s+(\S+)"
+        r"|(\S+)\s+ko\s+(\S+)\s+replace",
+        command,
+        re.IGNORECASE,
+    )
+    if m:
+        groups = [g for g in m.groups() if g is not None]
+        if len(groups) >= 2:
+            return {
+                "operation": "find_replace",
+                "column": col,
+                "old_value": groups[0],
+                "new_value": groups[1],
+            }
+
+    return None
+
+
+def _parse_rename(session_id: str, command: str, columns: list[str]) -> Optional[dict]:
+    """Extract current column and desired new name."""
+    col = _best_column(session_id, command, columns)
+    if col is None:
+        return None
+
+    m = re.search(
+        r"(?:naam|name)\s+(?:ko\s+)?(\S+)\s+(?:se\s+)?badlo"
+        r"|rename\s+\S+\s+to\s+(\S+)",
+        command,
+        re.IGNORECASE,
+    )
+    if m:
+        new_name = m.group(1) or m.group(2)
+        if new_name:
+            return {
+                "operation": "rename_column",
+                "column": col,
+                "new_name": new_name.strip("'\" "),
+            }
+    return None
+
+
+def _parse_cast(session_id: str, command: str, columns: list[str]) -> Optional[dict]:
+    """Extract column and target type for type casting."""
+    col = _best_column(session_id, command, columns)
+    if col is None:
+        return None
+
+    cmd_lower = command.lower()
+    target_dtype: str | None = None
+    if "int" in cmd_lower or "numeric" in cmd_lower or "number" in cmd_lower:
+        target_dtype = "Int64"
+    elif "float" in cmd_lower or "decimal" in cmd_lower:
+        target_dtype = "Float64"
+    elif "str" in cmd_lower or "string" in cmd_lower or "text" in cmd_lower:
+        target_dtype = "String"
+    elif "bool" in cmd_lower:
+        target_dtype = "Boolean"
+    elif "date" in cmd_lower or "datetime" in cmd_lower:
+        target_dtype = "Date"
+
+    if target_dtype:
+        return {"operation": "cast_type", "column": col, "target_dtype": target_dtype}
+    return None

core/parser/__init__.py

@@ -0,0 +1,3 @@
+"""
+Parser package.
+"""

core/parser/fallback.py

@@ -0,0 +1,164 @@
+"""
+Fallback parsing logic.
+Attempts fast regex-based parsing first, falling back to local heuristic/embedding parser,
+then to local Qwen GGUF model, and finally to the cloud Gemini LLM parser.
+"""
+from __future__ import annotations
+
+import os
+import json
+import urllib.request
+from typing import Any
+
+from core.intent_parser import parse_intent
+from core.parser.local_parser.parser import LocalIntentParser
+
+# Light weight high quality Qwen2.5 GGUF model url
+QWEN_GGUF_URL = "https://huggingface.co/Qwen/Qwen2.5-0.5B-Instruct-GGUF/resolve/main/qwen2.5-0.5b-instruct-q4_k_m.gguf"
+
+_local_parser: LocalIntentParser | None = None
+_qwen_model: Any | None = None
+
+def get_local_parser() -> LocalIntentParser:
+    """Lazily load the local heuristic & embedding parser."""
+    global _local_parser
+    if _local_parser is None:
+        _local_parser = LocalIntentParser()
+    return _local_parser
+
+def get_qwen_model(models_dir: str) -> Any | None:
+    """Lazily download and load the llama-cpp-python Qwen GGUF model."""
+    global _qwen_model
+    if _qwen_model is not None:
+        return _qwen_model
+
+    try:
+        from llama_cpp import Llama
+    except ImportError:
+        print("[Local Qwen] llama-cpp-python is not installed. Skipping local GGUF parser.")
+        return None
+
+    gguf_path = os.path.join(models_dir, "qwen2.5-0.5b-instruct-q4_k_m.gguf")
+    if not os.path.exists(gguf_path):
+        print(f"[Local Qwen] Downloading GGUF model to {gguf_path}...")
+        try:
+            os.makedirs(models_dir, exist_ok=True)
+            # Use urllib.request to avoid extra dependencies or cert issues
+            urllib.request.urlretrieve(QWEN_GGUF_URL, gguf_path)
+            print("[Local Qwen] GGUF model download complete.")
+        except Exception as e:
+            print(f"[Local Qwen] Failed to download GGUF model: {e}")
+            return None
+
+    try:
+        print(f"[Local Qwen] Loading model from {gguf_path}...")
+        # Load Qwen model with silent verbose logging and standard context window
+        _qwen_model = Llama(model_path=gguf_path, n_ctx=2048, verbose=False)
+        return _qwen_model
+    except Exception as e:
+        print(f"[Local Qwen] Error loading Llama model: {e}")
+        return None
+
+def parse_with_qwen(session_id: str, command: str) -> dict[str, Any] | None:
+    """Run local inference using Qwen GGUF via llama-cpp-python."""
+    from core.column_registry import column_registry
+    columns = column_registry.get_columns(session_id) or []
+
+    prompt = f"""You are a precise data engineering intent parser. Your task is to convert a user's natural language command into a structured JSON payload that represents a dataframe operation.
+
+Available Columns: {json.dumps(columns)}
+
+Your output must be a single valid JSON object. Choose the correct operation and populate the required keys. Do NOT wrap output in markdown formatting.
+
+Supported Operations & schemas:
+- "increase": {{"operation": "increase", "column": "<col>", "value": <val>, "is_percent": <bool>}}
+- "decrease": {{"operation": "decrease", "column": "<col>", "value": <val>, "is_percent": <bool>}}
+- "filter": {{"operation": "filter", "column": "<col>", "condition": "<op>", "filter_value": <val>}}
+- "sort_asc": {{"operation": "sort_asc", "column": "<col>"}}
+- "sort_desc": {{"operation": "sort_desc", "column": "<col>"}}
+- "sum": {{"operation": "sum", "column": "<col>"}}
+- "average": {{"operation": "average", "column": "<col>"}}
+- "count": {{"operation": "count", "column": "<col_or_null>"}}
+- "min": {{"operation": "min", "column": "<col>"}}
+- "max": {{"operation": "max", "column": "<col>"}}
+- "find_replace": {{"operation": "find_replace", "column": "<col>", "old_value": <val>, "new_value": <val>}}
+- "delete_column": {{"operation": "delete_column", "column": "<col>"}}
+- "rename_column": {{"operation": "rename_column", "column": "<col>", "new_name": "<str>"}}
+- "add_column": {{"operation": "add_column", "column": "<col_name>", "value": <val>}}
+- "remove_duplicates": {{"operation": "remove_duplicates", "column": "<col_or_null>"}}
+- "cast_type": {{"operation": "cast_type", "column": "<col>", "target_dtype": "<Int64|Float64|String|Boolean|Date>"}}
+
+Command: "{command}"
+JSON:"""
+
+    base_dir = os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))))
+    models_dir = os.path.join(base_dir, "zero-llm-engine", "models")
+    if not os.path.exists(models_dir):
+        try:
+            os.makedirs(models_dir, exist_ok=True)
+        except Exception:
+            models_dir = os.path.join("/tmp", "models")
+            os.makedirs(models_dir, exist_ok=True)
+
+    llm = get_qwen_model(models_dir)
+    if llm is None:
+        return None
+
+    try:
+        response = llm.create_chat_completion(
+            messages=[
+                {"role": "system", "content": "You are a precise data engineering intent parser. You output only raw, valid JSON."},
+                {"role": "user", "content": prompt}
+            ],
+            temperature=0.1,
+            max_tokens=256
+        )
+        content = response["choices"][0]["message"]["content"].strip()
+
+        # Clean markdown code block formatting if present
+        if content.startswith("```"):
+            lines = content.splitlines()
+            if lines[0].startswith("```"):
+                lines = lines[1:]
+            if lines and lines[-1].strip() == "```":
+                lines = lines[:-1]
+            content = "\n".join(lines).strip()
+
+        return json.loads(content)
+    except Exception as e:
+        print(f"[Local Qwen] GGUF Parsing failed or timed out: {e}")
+        return None
+
+def parse_intent_hybrid(session_id: str, command: str) -> dict | None:
+    """Parse user command using fast regex path, local orchestrator parser, local Qwen GGUF, or cloud Gemini."""
+    # 1. Fast Path: Regex / keyword resolution
+    fast_result = parse_intent(session_id, command)
+    if fast_result is not None:
+        print(f"[Hybrid Parser] Fast path success: {fast_result}")
+        return fast_result
+
+    # 2. Local Heuristic / Spelling / Synonym / Embeddings Parser
+    try:
+        local_parser = get_local_parser()
+        local_result = local_parser.parse_intent(session_id, command)
+        if local_result is not None:
+            confidence = local_result.pop("confidence", "low")
+            if confidence == "high":
+                print(f"[Hybrid Parser] Local parser high-confidence success: {local_result}")
+                return local_result
+            else:
+                print(f"[Hybrid Parser] Local parser yielded low confidence: {local_result}")
+    except Exception as e:
+        print(f"[Hybrid Parser] Exception in local parser: {e}")
+
+    # 3. Local Qwen GGUF Model Fallback
+    try:
+        qwen_result = parse_with_qwen(session_id, command)
+        if qwen_result is not None:
+            print(f"[Hybrid Parser] Local Qwen GGUF success: {qwen_result}")
+            return qwen_result
+    except Exception as e:
+        print(f"[Hybrid Parser] Exception in local Qwen fallback: {e}")
+
+    print("[Hybrid Parser] All local paths failed. Returning None.")
+    return None

core/parser/local_parser/__init__.py

@@ -0,0 +1,11 @@
+"""
+Local parser package.
+Provides offline, high-accuracy spelling correction, synonym mapping, ONNX embeddings column matching, and safe AST math evaluation.
+"""
+from __future__ import annotations
+
+from core.parser.local_parser.spelling import SymSpell
+from core.parser.local_parser.synonyms import SynonymMapper
+from core.parser.local_parser.embeddings import EmbeddingModel
+from core.parser.local_parser.ast_extractor import SafeMathEvaluator
+from core.parser.local_parser.parser import LocalIntentParser

core/parser/local_parser/ast_extractor.py

@@ -0,0 +1,107 @@
+"""
+Safe math evaluator using Python's ast library.
+Allows evaluation of mathematical and logical operations without exposing system to arbitrary code execution risks.
+"""
+from __future__ import annotations
+
+import ast
+import operator as op
+from typing import Any
+
+class SafeMathEvaluator:
+    """Safely evaluates mathematical and logical expressions using AST parsing."""
+    _operators = {
+        # Binary operators
+        ast.Add: op.add,
+        ast.Sub: op.sub,
+        ast.Mult: op.mul,
+        ast.Div: op.truediv,
+        ast.FloorDiv: op.floordiv,
+        ast.Mod: op.mod,
+        ast.Pow: op.pow,
+        # Unary operators
+        ast.USub: op.neg,
+        ast.UAdd: op.pos,
+        # Comparison operators
+        ast.Gt: op.gt,
+        ast.Lt: op.lt,
+        ast.GtE: op.ge,
+        ast.LtE: op.le,
+        ast.Eq: op.eq,
+        ast.NotEq: op.ne,
+        # Logical operators
+        ast.And: lambda a, b: a and b,
+        ast.Or: lambda a, b: a or b,
+        ast.Not: op.not_,
+    }
+
+    def __init__(self, variables: dict[str, Any] | None = None):
+        self.variables = variables or {}
+
+    def evaluate(self, expression: str) -> Any:
+        """Parse and evaluate a safe mathematical/logical expression."""
+        if not expression:
+            return None
+        try:
+            # Parse expression in 'eval' mode (expects a single expression)
+            tree = ast.parse(expression.strip(), mode="eval")
+            return self._eval(tree.body)
+        except Exception as e:
+            raise ValueError(f"Failed to evaluate expression '{expression}': {e}")
+
+    def _eval(self, node: ast.AST) -> Any:
+        """Recursively evaluate the AST node."""
+        if isinstance(node, ast.Num):  # Python < 3.8
+            return node.n
+        elif isinstance(node, ast.Constant):  # Python >= 3.8
+            return node.value
+        elif isinstance(node, ast.Name):
+            # Check variables dictionary
+            if node.id in self.variables:
+                return self.variables[node.id]
+            # Special boolean constants
+            if node.id == "True":
+                return True
+            if node.id == "False":
+                return False
+            if node.id == "None":
+                return None
+            raise NameError(f"Variable '{node.id}' is not defined or is not allowed.")
+        elif isinstance(node, ast.BinOp):
+            left = self._eval(node.left)
+            right = self._eval(node.right)
+            op_type = type(node.op)
+            if op_type in self._operators:
+                return self._operators[op_type](left, right)
+            raise TypeError(f"Unsupported binary operator: {op_type}")
+        elif isinstance(node, ast.UnaryOp):
+            operand = self._eval(node.operand)
+            op_type = type(node.op)
+            if op_type in self._operators:
+                return self._operators[op_type](operand)
+            raise TypeError(f"Unsupported unary operator: {op_type}")
+        elif isinstance(node, ast.Compare):
+            left = self._eval(node.left)
+            for operation, comparator in zip(node.ops, node.comparators):
+                right = self._eval(comparator)
+                op_type = type(operation)
+                if op_type in self._operators:
+                    if not self._operators[op_type](left, right):
+                        return False
+                    left = right
+                else:
+                    raise TypeError(f"Unsupported comparison operator: {op_type}")
+            return True
+        elif isinstance(node, ast.BoolOp):
+            op_type = type(node.op)
+            if op_type not in self._operators:
+                raise TypeError(f"Unsupported logical operator: {op_type}")
+            values = [self._eval(val) for val in node.values]
+            if not values:
+                return False
+            result = values[0]
+            for val in values[1:]:
+                result = self._operators[op_type](result, val)
+            return result
+        else:
+            raise TypeError(f"Unsupported expression node: {type(node)}")

core/parser/local_parser/embeddings.py

@@ -0,0 +1,205 @@
+"""
+Embedding module using a quantized ONNX MiniLM model and a pure Python WordPiece tokenizer.
+Calculates cosine similarity and caches column embeddings for performance.
+"""
+from __future__ import annotations
+
+import os
+import urllib.request
+import numpy as np
+import onnxruntime as ort
+import re
+from typing import Sequence
+
+# Default Hugging Face URLs for the quantized all-MiniLM-L6-v2 model and vocab
+MODEL_URL = "https://huggingface.co/onnx-community/all-MiniLM-L6-v2-ONNX/resolve/main/onnx/model_quantized.onnx"
+VOCAB_URL = "https://huggingface.co/onnx-community/all-MiniLM-L6-v2-ONNX/resolve/main/vocab.txt"
+
+class WordPieceTokenizer:
+    """Pure Python implementation of a WordPiece tokenizer."""
+    def __init__(self, vocab_path: str):
+        self.vocab: dict[str, int] = {}
+        with open(vocab_path, "r", encoding="utf-8") as f:
+            for i, line in enumerate(f):
+                token = line.strip()
+                self.vocab[token] = i
+
+        self.unk_token = "[UNK]"
+        self.cls_token = "[CLS]"
+        self.sep_token = "[SEP]"
+
+        self.unk_id = self.vocab.get(self.unk_token, 100)
+        self.cls_id = self.vocab.get(self.cls_token, 101)
+        self.sep_id = self.vocab.get(self.sep_token, 102)
+
+    def tokenize_word(self, word: str) -> list[str]:
+        """Tokenize a single word into WordPiece subwords."""
+        if word in self.vocab:
+            return [word]
+
+        tokens = []
+        start = 0
+        is_bad = False
+        while start < len(word):
+            end = len(word)
+            cur_substr = None
+            while start < end:
+                substr = word[start:end]
+                if start > 0:
+                    substr = "##" + substr
+                if substr in self.vocab:
+                    cur_substr = substr
+                    break
+                end -= 1
+            if cur_substr is None:
+                is_bad = True
+                break
+            tokens.append(cur_substr)
+            start = end
+
+        if is_bad:
+            return [self.unk_token]
+        return tokens
+
+    def encode(self, text: str, max_length: int = 128) -> dict[str, np.ndarray]:
+        """Encode text into model inputs (input_ids, attention_mask, token_type_ids)."""
+        text = text.lower()
+        # Basic word and punctuation splitter
+        words = re.findall(r"\w+|[^\w\s]", text, re.UNICODE)
+        tokens = []
+        for word in words:
+            tokens.extend(self.tokenize_word(word))
+
+        # Truncate
+        if len(tokens) > max_length - 2:
+            tokens = tokens[:max_length - 2]
+
+        # Build token IDs
+        input_ids = [self.cls_id] + [self.vocab.get(t, self.unk_id) for t in tokens] + [self.sep_id]
+        attention_mask = [1] * len(input_ids)
+        token_type_ids = [0] * len(input_ids)
+
+        # Pad to max_length
+        padding_len = max_length - len(input_ids)
+        if padding_len > 0:
+            input_ids.extend([0] * padding_len)
+            attention_mask.extend([0] * padding_len)
+            token_type_ids.extend([0] * padding_len)
+
+        # Convert to numpy arrays of type int64 (as expected by ONNX model)
+        return {
+            "input_ids": np.array([input_ids], dtype=np.int64),
+            "attention_mask": np.array([attention_mask], dtype=np.int64),
+            "token_type_ids": np.array([token_type_ids], dtype=np.int64),
+        }
+
+def cosine_similarity(v1: np.ndarray, v2: np.ndarray) -> float:
+    """Calculate the cosine similarity between two 1D vectors."""
+    dot = np.dot(v1, v2)
+    norm1 = np.linalg.norm(v1)
+    norm2 = np.linalg.norm(v2)
+    if norm1 == 0 or norm2 == 0:
+        return 0.0
+    return float(dot / (norm1 * norm2))
+
+class EmbeddingModel:
+    """ONNX-based text embedding generator with built-in WordPiece tokenization."""
+    def __init__(self, cache_dir: str | None = None):
+        if cache_dir is None:
+            # First try zero-llm-engine/models directory, fallback to temp dir /tmp/models
+            base_dir = os.path.dirname(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))))
+            cache_dir = os.path.join(base_dir, "zero-llm-engine", "models", "onnx_cache")
+            if not os.path.exists(cache_dir):
+                try:
+                    os.makedirs(cache_dir, exist_ok=True)
+                except Exception:
+                    # Fallback if the folder is read-only (e.g. some system settings or Docker environments)
+                    cache_dir = os.path.join("/tmp", "onnx_cache")
+                    os.makedirs(cache_dir, exist_ok=True)
+
+        self.cache_dir = cache_dir
+        self.model_path = os.path.join(cache_dir, "model_quantized.onnx")
+        self.vocab_path = os.path.join(cache_dir, "vocab.txt")
+        self.session: ort.InferenceSession | None = None
+        self.tokenizer: WordPieceTokenizer | None = None
+        
+        # In-memory embedding cache: maps text string -> np.ndarray embedding
+        self._embedding_cache: dict[str, np.ndarray] = {}
+
+    def ensure_model_files(self) -> None:
+        """Download model and vocabulary files if they do not exist locally."""
+        if not os.path.exists(self.vocab_path):
+            print(f"[Embeddings] Downloading vocabulary to {self.vocab_path}...")
+            urllib.request.urlretrieve(VOCAB_URL, self.vocab_path)
+        if not os.path.exists(self.model_path):
+            print(f"[Embeddings] Downloading quantized ONNX model to {self.model_path}...")
+            urllib.request.urlretrieve(MODEL_URL, self.model_path)
+
+    def load_model(self) -> None:
+        """Ensure files are downloaded and load the ONNX session and tokenizer."""
+        if self.session is not None and self.tokenizer is not None:
+            return
+            
+        self.ensure_model_files()
+        
+        # Initialize tokenization & ONNX session
+        self.tokenizer = WordPieceTokenizer(self.vocab_path)
+        # Using CPU execution provider by default for maximum compatibility
+        self.session = ort.InferenceSession(self.model_path, providers=["CPUExecutionProvider"])
+
+    def get_embedding(self, text: str) -> np.ndarray:
+        """Generate a 1D mean-pooled normalized embedding vector for the text."""
+        self.load_model()
+        text_key = text.lower().strip()
+        if text_key in self._embedding_cache:
+            return self._embedding_cache[text_key]
+
+        assert self.tokenizer is not None
+        assert self.session is not None
+
+        # Tokenize and format inputs
+        inputs = self.tokenizer.encode(text_key)
+        
+        # Run ONNX inference
+        outputs = self.session.run(None, inputs)
+        # The first output contains the token embeddings [batch_size, seq_len, hidden_dim]
+        token_embeddings = outputs[0]
+        attention_mask = inputs["attention_mask"]
+
+        # Perform mean pooling over the active tokens
+        input_mask_expanded = np.expand_dims(attention_mask, axis=-1)
+        input_mask_expanded = np.broadcast_to(input_mask_expanded, token_embeddings.shape)
+        
+        sum_embeddings = np.sum(token_embeddings * input_mask_expanded, axis=1)
+        sum_mask = np.clip(np.sum(input_mask_expanded, axis=1), a_min=1e-9, a_max=None)
+        
+        # Calculate mean pooled embedding (vector shape: [hidden_dim])
+        mean_pooled = (sum_embeddings / sum_mask)[0]
+        
+        # L2 Normalize
+        norm = np.linalg.norm(mean_pooled)
+        if norm > 0:
+            mean_pooled = mean_pooled / norm
+            
+        self._embedding_cache[text_key] = mean_pooled
+        return mean_pooled
+
+    def match_column(self, text: str, columns: Sequence[str], threshold: float = 0.4) -> tuple[str | None, float]:
+        """Match query text to the best column name using cosine similarity."""
+        if not columns:
+            return None, 0.0
+
+        query_emb = self.get_embedding(text)
+        best_col = None
+        best_sim = -1.0
+
+        for col in columns:
+            col_emb = self.get_embedding(col)
+            sim = cosine_similarity(query_emb, col_emb)
+            if sim > best_sim:
+                best_sim = sim
+                best_col = col
+
+        if best_sim >= threshold:
+            return best_col, best_sim
+        return None, best_sim

core/parser/local_parser/parser.py

@@ -0,0 +1,303 @@
+"""
+Orchestration parser module.
+Integrates spelling correction, synonym mapping, ONNX embeddings matching, and parameter extraction.
+"""
+from __future__ import annotations
+
+import re
+from typing import Any
+from core.column_registry import column_registry
+from core.parser.local_parser.spelling import SymSpell
+from core.parser.local_parser.synonyms import SynonymMapper
+from core.parser.local_parser.embeddings import EmbeddingModel
+from core.parser.local_parser.ast_extractor import SafeMathEvaluator
+
+# Standard operation keywords (from intent_parser.py)
+OPERATION_KEYWORDS: dict[str, list[str]] = {
+    "increase": ["badhao", "badha do", "increase", "barha do", "barhao", "zyada karo", "bada karo", "grow", "raise", "badha dijiye"],
+    "decrease": ["ghatao", "ghata do", "decrease", "kam karo", "kam kar do", "chhota karo", "reduce", "cut", "minus karo", "ghata dijiye"],
+    "filter": ["sirf", "only", "filter", "dikhao", "show only", "show me", "bas", "wale dikhao", "where", "jitne", "laao"],
+    "sort_asc": ["chota se bada", "ascending", "a to z", "low to high", "smallest first", "ascending order", "a-z", "asc"],
+    "sort_desc": ["bada se chota", "descending", "z to a", "high to low", "largest first", "descending order", "z-a", "desc", "bade se chhote"],
+    "sum": ["sum", "total", "jod", "yog", "add up", "total batao", "kul", "jama"],
+    "average": ["average", "avg", "mean", "samanya", "average nikalo"],
+    "count": ["count", "ginti", "kitne", "kitni rows", "count karo", "kitni", "rows kitne", "total rows"],
+    "min": ["minimum", "min", "sabse chhota", "lowest", "kam se kam"],
+    "max": ["maximum", "max", "sabse bada", "highest", "zyada se zyada"],
+    "find_replace": ["replace", "badlo", "change", "find", "dhundho", "substitute", "replace karo", "change karo", "naye se badlo"],
+    "delete_column": ["delete column", "column hatao", "column delete karo", "remove column", "column remove karo", "column hatado", "column drop karo"],
+    "rename_column": ["rename column", "column ka naam badlo", "column rename karo", "name change karo", "naam badlo", "rename karo"],
+    "add_column": ["add column", "naya column banao", "column add karo", "new column", "column create karo"],
+    "remove_duplicates": ["duplicate hatao", "duplicates remove karo", "unique rakho", "duplicate remove", "repeat hatao"],
+    "cast_type": ["type badlo", "data type change", "convert type", "type convert karo", "numeric banao", "string banao"],
+}
+
+class LocalIntentParser:
+    """Orchestrates local heuristic parsing using spelling correction, synonyms, embeddings, and AST."""
+    def __init__(self):
+        self.synonym_mapper = SynonymMapper()
+        self.embedding_model = EmbeddingModel()
+        self.math_evaluator = SafeMathEvaluator()
+        
+        # Load spelling corrector with keywords
+        self.sym_spell = SymSpell(max_edit_distance=2)
+        for op_list in OPERATION_KEYWORDS.values():
+            for kw in op_list:
+                # Add word tokens to spelling index
+                for token in re.findall(r'[a-zA-Z]+', kw):
+                    self.sym_spell.add_word(token)
+
+    def _get_columns(self, session_id: str) -> list[str]:
+        return column_registry.get_columns(session_id) or []
+
+    def parse_intent(self, session_id: str, command: str) -> dict[str, Any] | None:
+        """Parse natural language command into structured dataframe operation JSON."""
+        columns = self._get_columns(session_id)
+        if not columns:
+            return None
+
+        # 1. Spelling correction
+        # Build a spelling corrector with current column names dynamically included
+        local_sym_spell = SymSpell(max_edit_distance=2)
+        # copy keywords
+        for w in self.sym_spell.words:
+            local_sym_spell.add_word(w)
+        # Add column names
+        for col in columns:
+            local_sym_spell.add_word(col)
+            for token in re.findall(r'[a-zA-Z]+', col):
+                local_sym_spell.add_word(token)
+
+        corrected_cmd = local_sym_spell.correct_query(command)
+
+        # 2. Synonym mapping and Hinglish normalization
+        norm_cmd = self.synonym_mapper.normalize_text(corrected_cmd)
+
+        # 3. Match operations
+        operation = self._match_operation(norm_cmd)
+        if not operation:
+            return None
+
+        # 4. Column matching (Exact or Semantic)
+        column_match, confidence = self._resolve_column(norm_cmd, columns)
+
+        # 5. Parameter extraction based on matched operation
+        result: dict[str, Any] = {"operation": operation}
+
+        if operation in ["remove_duplicates", "count"]:
+            # Column is optional for remove_duplicates and count
+            if column_match:
+                result["column"] = column_match
+            else:
+                result["column"] = None
+            result["confidence"] = "high"
+            return result
+
+        if operation == "delete_column":
+            if not column_match:
+                return None
+            result["column"] = column_match
+            result["confidence"] = "high"
+            return result
+
+        if operation == "rename_column":
+            # Extract new name from patterns like "rename X to Y"
+            rename_match = re.search(r"rename\s+(?:column\s+)?(\w+)\s+(?:to|as)\s+(\w+)", norm_cmd, re.IGNORECASE)
+            if rename_match:
+                old_name_cand = rename_match.group(1)
+                new_name = rename_match.group(2)
+                # Resolve old name using columns list
+                resolved_old, _ = self._resolve_column(old_name_cand, columns)
+                result["column"] = resolved_old or column_match
+                result["new_name"] = new_name
+                result["confidence"] = "high" if result["column"] else "low"
+                return result
+            # Try fallback: split by "to" or "as"
+            parts = re.split(r"\b(?:to|as)\b", norm_cmd)
+            if len(parts) >= 2:
+                new_name = parts[-1].strip().split()[-1]
+                result["column"] = column_match
+                result["new_name"] = new_name
+                result["confidence"] = "high" if column_match else "low"
+                return result
+            return None
+
+        if operation == "find_replace":
+            # Look for patterns: "replace A with B"
+            replace_match = re.search(r"replace\s+(.+?)\s+with\s+(.+)", norm_cmd, re.IGNORECASE)
+            if replace_match:
+                old_val = replace_match.group(1).strip()
+                new_val = replace_match.group(2).strip()
+                # Remove column name references from old_val if present
+                if column_match and old_val.startswith(column_match.lower()):
+                    old_val = old_val[len(column_match):].strip()
+                
+                result["column"] = column_match
+                result["old_value"] = self._try_parse_numeric(old_val)
+                result["new_value"] = self._try_parse_numeric(new_val)
+                result["confidence"] = "high" if column_match else "low"
+                return result
+            return None
+
+        if operation == "add_column":
+            # Patterns: "add column X with value Y", "new column X = Y"
+            add_match = re.search(r"(?:add|new)\s+column\s+(\w+)(?:\s+(?:with|value|=)\s+(.+))?", norm_cmd, re.IGNORECASE)
+            if add_match:
+                col_name = add_match.group(1)
+                default_val_str = add_match.group(2)
+                default_val = self._try_parse_numeric(default_val_str) if default_val_str else None
+                result["column"] = col_name
+                result["value"] = default_val
+                result["confidence"] = "high"
+                return result
+            return None
+
+        if operation == "cast_type":
+            # Check target datatype
+            target_dtype = self._resolve_dtype(norm_cmd)
+            if not target_dtype:
+                return None
+            result["column"] = column_match
+            result["target_dtype"] = target_dtype
+            result["confidence"] = "high" if column_match else "low"
+            return result
+
+        if operation in ["increase", "decrease"]:
+            if not column_match:
+                return None
+            # Check if percentage
+            is_percent = "%" in command or "percent" in norm_cmd
+            # Find numbers
+            num_match = re.search(r"(\d+(?:\.\d+)?)", norm_cmd)
+            value = float(num_match.group(1)) if num_match else 0.0
+            
+            result["column"] = column_match
+            result["value"] = value
+            result["is_percent"] = is_percent
+            result["confidence"] = "high"
+            return result
+
+        if operation == "filter":
+            if not column_match:
+                return None
+            
+            # Resolve comparison operator
+            condition = "=="
+            for op_sym in [">=", "<=", ">", "<", "!=", "=="]:
+                if op_sym in norm_cmd:
+                    condition = op_sym
+                    break
+            else:
+                if "contains" in norm_cmd or "like" in norm_cmd:
+                    condition = "contains"
+                elif "equal" in norm_cmd:
+                    condition = "=="
+                elif "greater" in norm_cmd:
+                    condition = ">"
+                elif "less" in norm_cmd:
+                    condition = "<"
+
+            # Try to extract filter value
+            # Split query by operator or column name to isolate the value
+            filter_val_str = ""
+            if condition in norm_cmd:
+                parts = norm_cmd.split(condition, 1)
+                if len(parts) == 2:
+                    filter_val_str = parts[1].strip()
+            else:
+                # Fallback to finding numeric or text token at the end
+                words = norm_cmd.split()
+                if words:
+                    filter_val_str = words[-1]
+
+            # Clean filter val string
+            filter_val_str = re.sub(r"\b(?:karo|dikhao|bas|only|show|hai|hain)\b", "", filter_val_str).strip()
+            # Remove any trailing periods
+            filter_val_str = filter_val_str.rstrip(".")
+
+            filter_value = self._try_parse_numeric(filter_val_str)
+            
+            result["column"] = column_match
+            result["condition"] = condition
+            result["filter_value"] = filter_value
+            result["confidence"] = "high"
+            return result
+
+        # Standard aggregation / simple operations (sum, average, min, max, sort_asc, sort_desc)
+        if not column_match:
+            return None
+            
+        result["column"] = column_match
+        result["confidence"] = "high" if confidence >= 0.5 else "low"
+        return result
+
+    def _match_operation(self, command: str) -> str | None:
+        """Choose the operation with the highest keyword match score."""
+        cmd = command.lower()
+        scores: dict[str, int] = {}
+        for op, keywords in OPERATION_KEYWORDS.items():
+            for kw in keywords:
+                if kw in cmd:
+                    # Weight by keyword length to favor more specific matches
+                    scores[op] = scores.get(op, 0) + len(kw)
+        if not scores:
+            return None
+        return max(scores, key=scores.get)
+
+    def _resolve_column(self, command: str, columns: list[str]) -> tuple[str | None, float]:
+        """Match the query to a column name, supporting exact and semantic resolution."""
+        cmd_lower = command.lower()
+        
+        # 1. Exact / Substring Match (Case-insensitive)
+        for col in columns:
+            if col.lower() in cmd_lower:
+                return col, 1.0
+
+        # 2. Semantic Similarity Fallback
+        # Extract keywords to reduce noise in the query string
+        clean_text = cmd_lower
+        for op_list in OPERATION_KEYWORDS.values():
+            for kw in op_list:
+                clean_text = re.sub(rf'\b{re.escape(kw)}\b', "", clean_text)
+                
+        # Clean extra spaces
+        clean_text = " ".join(clean_text.split())
+        if not clean_text:
+            clean_text = cmd_lower
+
+        try:
+            return self.embedding_model.match_column(clean_text, columns, threshold=0.4)
+        except Exception as e:
+            print(f"[Local Parser] Semantic matching failed: {e}")
+            # Fallback to first column or None
+            return None, 0.0
+
+    def _try_parse_numeric(self, val_str: str) -> Any:
+        """Helper to cast string to int or float if applicable, strip quotes if string."""
+        val_str = val_str.strip().strip("'\"")
+        try:
+            if "." in val_str:
+                return float(val_str)
+            return int(val_str)
+        except ValueError:
+            # Check boolean values
+            if val_str.lower() == "true":
+                return True
+            if val_str.lower() == "false":
+                return False
+            return val_str
+
+    def _resolve_dtype(self, command: str) -> str | None:
+        """Match string to target datatype name."""
+        cmd = command.lower()
+        if "int" in cmd or "integer" in cmd or "numeric" in cmd or "number" in cmd:
+            return "Int64"
+        if "float" in cmd or "double" in cmd or "decimal" in cmd:
+            return "Float64"
+        if "string" in cmd or "text" in cmd or "character" in cmd:
+            return "String"
+        if "bool" in cmd or "boolean" in cmd:
+            return "Boolean"
+        if "date" in cmd or "time" in cmd:
+            return "Date"
+        return None

core/parser/local_parser/spelling.py

@@ -0,0 +1,109 @@
+"""
+Spelling correction module using SymSpell (Symmetric Delete Spelling Correction).
+Optimized for O(1) delete lookups to match column names and operation keywords.
+"""
+from __future__ import annotations
+
+import re
+
+def _get_deletes(word: str, max_edit_distance: int = 2) -> set[str]:
+    """Generate deletes for a word up to a maximum edit distance."""
+    deletes = set()
+    queue = {word}
+    for _ in range(max_edit_distance):
+        next_queue = set()
+        for w in queue:
+            if len(w) > 1:
+                for i in range(len(w)):
+                    del_w = w[:i] + w[i+1:]
+                    deletes.add(del_w)
+                    next_queue.add(del_w)
+        queue = next_queue
+    return deletes
+
+def _lev_dist(s1: str, s2: str) -> int:
+    """Calculate the Levenshtein distance between two strings."""
+    if len(s1) < len(s2):
+        return _lev_dist(s2, s1)
+    if len(s2) == 0:
+        return len(s1)
+    previous_row = list(range(len(s2) + 1))
+    for i, c1 in enumerate(s1):
+        current_row = [i + 1]
+        for j, c2 in enumerate(s2):
+            insertions = previous_row[j + 1] + 1
+            deletions = current_row[j] + 1
+            substitutions = previous_row[j] + (c1 != c2)
+            current_row.append(min(insertions, deletions, substitutions))
+        previous_row = current_row
+    return previous_row[-1]
+
+class SymSpell:
+    """A lightweight symmetric delete spelling corrector."""
+    def __init__(self, max_edit_distance: int = 2):
+        self.max_edit_distance = max_edit_distance
+        # Maps delete_item -> set of original_words
+        self.deletes: dict[str, set[str]] = {}
+        self.words: set[str] = set()
+
+    def add_word(self, word: str) -> None:
+        """Index a word and its deletes for O(1) spelling lookup."""
+        word = word.lower().strip()
+        if not word or word in self.words:
+            return
+        self.words.add(word)
+        
+        # Index word itself
+        if word not in self.deletes:
+            self.deletes[word] = set()
+        self.deletes[word].add(word)
+        
+        # Index deletions
+        for delete in _get_deletes(word, self.max_edit_distance):
+            if delete not in self.deletes:
+                self.deletes[delete] = set()
+            self.deletes[delete].add(word)
+
+    def lookup(self, word: str) -> list[str]:
+        """Find candidate words matching the spelling of input word."""
+        word = word.lower().strip()
+        if not word:
+            return []
+        if word in self.words:
+            return [word]
+        
+        candidates: set[str] = set()
+        # 1. Direct delete match
+        if word in self.deletes:
+            candidates.update(self.deletes[word])
+            
+        # 2. Deletes of word match
+        for delete in _get_deletes(word, self.max_edit_distance):
+            if delete in self.deletes:
+                candidates.update(self.deletes[delete])
+            if delete in self.words:
+                candidates.add(delete)
+                
+        # Score and rank candidates by Levenshtein distance
+        scored = []
+        for cand in candidates:
+            dist = _lev_dist(word, cand)
+            if dist <= self.max_edit_distance:
+                scored.append((cand, dist))
+                
+        # Sort by distance first, then length (longer words first for ties)
+        scored.sort(key=lambda x: (x[1], -len(x[0])))
+        return [c for c, _ in scored]
+
+    def correct_query(self, query: str) -> str:
+        """Correct typos in words within the query string."""
+        # Find all alphabet-only words
+        words = re.findall(r'[a-zA-Z]+', query)
+        corrected = query
+        for w in words:
+            if len(w) > 2:  # Only correct words longer than 2 characters
+                suggestions = self.lookup(w)
+                if suggestions:
+                    # Match exact word boundary to prevent partial replacements
+                    corrected = re.sub(rf'\b{re.escape(w)}\b', suggestions[0], corrected, flags=re.IGNORECASE)
+        return corrected

core/parser/local_parser/synonyms.py

@@ -0,0 +1,69 @@
+"""
+Synonyms and Hinglish normalization module.
+Maps Hinglish phrases and synonyms to canonical DataFrame operations and column concepts.
+"""
+from __future__ import annotations
+
+import re
+
+# Hinglish to English operation mappings
+HINGLISH_TO_ENGLISH_OP: dict[str, str] = {
+    "badhao": "increase", "badha do": "increase", "barha do": "increase", "barhao": "increase",
+    "zyada karo": "increase", "bada karo": "increase", "badha dijiye": "increase",
+    "ghatao": "decrease", "ghata do": "decrease", "kam karo": "decrease", "kam kar do": "decrease",
+    "chhota karo": "decrease", "reduce": "decrease", "cut": "decrease", "minus karo": "decrease", "ghata dijiye": "decrease",
+    "sirf": "filter", "dikhao": "filter", "show only": "filter", "show me": "filter",
+    "bas": "filter", "wale dikhao": "filter", "jitne": "filter", "laao": "filter",
+    "chota se bada": "sort_asc", "ascending": "sort_asc", "low to high": "sort_asc", "smallest first": "sort_asc",
+    "bada se chota": "sort_desc", "descending": "sort_desc", "high to low": "sort_desc", "largest first": "sort_desc",
+    "jod": "sum", "yog": "sum", "add up": "sum", "total batao": "sum", "kul": "sum", "jama": "sum",
+    "samanya": "average", "average nikalo": "average", "avg": "average",
+    "ginti": "count", "kitne": "count", "kitni rows": "count", "count karo": "count", "kitni": "count",
+    "sabse chhota": "min", "lowest": "min", "kam se kam": "min",
+    "sabse bada": "max", "highest": "max", "zyada se zyada": "max",
+    "badlo": "find_replace", "change": "find_replace", "dhundho": "find_replace", "replace karo": "find_replace",
+    "column hatao": "delete_column", "column delete karo": "delete_column", "remove column": "delete_column",
+    "column ka naam badlo": "rename_column", "column rename karo": "rename_column", "naam badlo": "rename_column",
+    "naya column banao": "add_column", "column add karo": "add_column", "new column": "add_column",
+    "duplicate hatao": "remove_duplicates", "duplicates remove karo": "remove_duplicates", "unique rakho": "remove_duplicates",
+    "type badlo": "cast_type", "data type change": "cast_type", "convert type": "cast_type",
+}
+
+# Common column name synonym mappings
+COMMON_COLUMN_SYNONYMS: dict[str, str] = {
+    "vetan": "salary", "kamai": "salary", "paisa": "salary", "income": "salary",
+    "umar": "age", "umra": "age",
+    "naam": "name",
+    "mulya": "price", "daam": "price", "keemat": "price", "cost": "price",
+    "tareekh": "date", "din": "date",
+    "shahar": "city", "shehar": "city",
+    "desh": "country",
+    "phone": "mobile", "mobile number": "mobile", "contact": "mobile",
+}
+
+class SynonymMapper:
+    """Handles mapping of synonym phrases and normalizes Hinglish queries."""
+    def __init__(self, custom_column_synonyms: dict[str, str] | None = None):
+        self.op_map = HINGLISH_TO_ENGLISH_OP
+        self.col_map = {**COMMON_COLUMN_SYNONYMS, **(custom_column_synonyms or {})}
+
+    def normalize_text(self, text: str) -> str:
+        """Normalize general Hinglish operations and column names to canonical terms."""
+        text_lower = text.lower().strip()
+        
+        # 1. Normalize operations (longer/more specific phrases first to prevent partial match issues)
+        sorted_ops = sorted(self.op_map.keys(), key=len, reverse=True)
+        for h_op in sorted_ops:
+            e_op = self.op_map[h_op]
+            # Replace complete word/phrase boundaries where possible
+            if h_op in text_lower:
+                text_lower = re.sub(rf'\b{re.escape(h_op)}\b', e_op, text_lower)
+                
+        # 2. Normalize columns
+        sorted_cols = sorted(self.col_map.keys(), key=len, reverse=True)
+        for h_col in sorted_cols:
+            e_col = self.col_map[h_col]
+            if h_col in text_lower:
+                text_lower = re.sub(rf'\b{re.escape(h_col)}\b', e_col, text_lower)
+                
+        return text_lower

core/router.py

@@ -0,0 +1,56 @@
+"""
+Router — maps a validated intent dict to the correct tool function.
+
+Pure dict lookup, zero overhead. Each tool function has the signature:
+    tool(session_id: str, intent: dict) -> dict
+and returns {"message": str, "diff": dict}.
+"""
+from __future__ import annotations
+
+from typing import Callable, Optional
+
+# Import all tools — they register themselves via TOOL_REGISTRY below
+from core.a_to_z.student.update import execute_increase, execute_decrease
+from core.a_to_z.student.filter_tool import execute_filter
+from core.a_to_z.student.sort_tool import execute_sort
+from core.a_to_z.student.aggregate import execute_sum, execute_average, execute_count, execute_min, execute_max
+from core.a_to_z.student.column_ops import execute_delete_column, execute_rename_column, execute_add_column, execute_cast_type
+from core.a_to_z.business.find_replace import execute_find_replace
+from core.a_to_z.business.dedup import execute_remove_duplicates
+
+
+TOOL_MAP: dict[str, Callable[[str, dict], dict]] = {
+    # Update
+    "increase": execute_increase,
+    "decrease": execute_decrease,
+    # Filter & Sort
+    "filter": execute_filter,
+    "sort_asc": execute_sort,
+    "sort_desc": execute_sort,
+    # Aggregate
+    "sum": execute_sum,
+    "average": execute_average,
+    "count": execute_count,
+    "min": execute_min,
+    "max": execute_max,
+    # Column operations
+    "delete_column": execute_delete_column,
+    "rename_column": execute_rename_column,
+    "add_column": execute_add_column,
+    "cast_type": execute_cast_type,
+    # Data cleaning
+    "find_replace": execute_find_replace,
+    "remove_duplicates": execute_remove_duplicates,
+}
+
+
+def dispatch(session_id: str, intent: dict) -> dict:
+    """Route intent to the matching tool. Returns tool's result dict."""
+    op = intent["operation"]
+    tool_fn = TOOL_MAP.get(op)
+    if tool_fn is None:
+        return {
+            "message": f"'{op}' operation supported nahi hai",
+            "diff": {},
+        }
+    return tool_fn(session_id, intent)

core/validator.py

@@ -0,0 +1,71 @@
+"""
+Pre-execution validator.
+
+Checks that a resolved intent is compatible with the dataframe schema
+before any data gets touched. This is the safety net that stops
+"increase the name column" from ever reaching Polars.
+"""
+from __future__ import annotations
+
+from typing import Optional
+import polars as pl
+
+from config import DATA_DIR
+from core.column_registry import column_registry
+
+
+# Operations that REQUIRE numeric dtype on the target column
+NUMERIC_ONLY_OPS = {"increase", "decrease", "sum", "average", "min", "max"}
+
+
+def get_schema(session_id: str) -> Optional[pl.Schema]:
+    """Read just the Parquet metadata to get column dtypes — no data loaded."""
+    import os
+    from services.session_manager import session_manager
+    pq_path = session_manager.get_filepath(session_id)
+    if not os.path.exists(pq_path):
+        return None
+    try:
+        schema = pl.scan_parquet(pq_path).collect_schema()
+        return schema
+    except Exception:
+        return None
+
+
+def validate_intent(session_id: str, intent: dict) -> Optional[str]:
+    """Return an error string if the intent is invalid, else ``None``."""
+    schema = get_schema(session_id)
+    if schema is None:
+        return "Session ka data nahi mila — dobara upload karo"
+
+    col = intent.get("column")
+    op = intent.get("operation", "")
+
+    # ── Column existence (skip for operations that don't target a column) ──
+    no_column_ops = {"remove_duplicates"}
+    if op not in no_column_ops and col:
+        if col not in schema:
+            return f"'{col}' column nahi hai file mein"
+
+    # ── Numeric dtype check ────────────────────────────────────────────
+    if op in NUMERIC_ONLY_OPS and col:
+        dtype = schema[col]
+        if dtype not in (pl.Float64, pl.Int64, pl.Int32, pl.Float32, pl.UInt64, pl.UInt32):
+            return f"'{col}' numeric column nahi hai ({dtype}), ye operation apply nahi ho sakta"
+
+    # ── Filter: need condition + value ──────────────────────────────────
+    if op == "filter":
+        if intent.get("condition") is None or intent.get("filter_value") is None:
+            return "Filter ke liye condition aur value dono chahiye, jaise: 'salary > 50000 dikhao'"
+
+    # ── Find & Replace: need old + new ──────────────────────────────────
+    if op == "find_replace":
+        if not intent.get("old_value") or not intent.get("new_value"):
+            return "Find & Replace ke liye purana aur naya value dono chahiye, jaise: 'Active ko Inactive se badlo'"
+
+    # ── Rename: need new_name ───────────────────────────────────────────
+    if op == "rename_column":
+        if not intent.get("new_name"):
+            return "Rename ke liye naya naam chahiye, jaise: 'name ko full_name rename karo'"
+
+    return None  # All good

main.py

@@ -0,0 +1,66 @@
+"""
+Zero-LLM Multi-Agent Excel/CSV Editing Engine
+==============================================
+
+Production-ready FastAPI server.  No LLM call anywhere.
+
+Usage:
+    uvicorn main:app --host 0.0.0.0 --port 8000 --reload
+
+Endpoints:
+    POST   /api/upload                Upload CSV/Excel/Parquet
+    GET    /api/download/{session_id}  Download as CSV
+    GET    /api/session/{session_id}   Session metadata
+    DELETE /api/session/{session_id}   Delete session
+    GET    /api/history/{session_id}   Command history
+    WS     /ws/{session_id}            Real-time command channel
+"""
+from __future__ import annotations
+
+from contextlib import asynccontextmanager
+from fastapi import FastAPI
+from fastapi.middleware.cors import CORSMiddleware
+
+from api.routes import router as rest_router
+from api.websocket import websocket_handler
+from services.audit_service import close as close_audit
+
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    """Startup / shutdown hooks."""
+    yield
+    await close_audit()
+
+
+app = FastAPI(
+    title="Zero-LLM Data Engine",
+    description="Natural-language Excel/CSV editing — no LLM, pure Python + Polars",
+    version="1.0.0",
+    lifespan=lifespan,
+)
+
+# CORS — allow all origins for dev; restrict in production
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+# Mount REST routes under /api
+app.include_router(rest_router, prefix="/api")
+
+# WebSocket endpoint
+from fastapi import WebSocket
+@app.websocket("/ws/{session_id}")
+async def ws_endpoint(ws: WebSocket, session_id: str):
+    await websocket_handler(ws, session_id)
+
+
+# ── Health check ────────────────────────────────────────────────────
+
+@app.get("/", tags=["health"])
+async def root():
+    return {"status": "running", "engine": "zero-llm", "version": "1.0.0"}

models/__init__.py

@@ -0,0 +1 @@
+# models package

models/schemas.py

@@ -0,0 +1,43 @@
+"""
+Pydantic schemas for request / response payloads.
+"""
+from __future__ import annotations
+
+from typing import Optional, Any
+from pydantic import BaseModel
+
+
+# ── Requests ────────────────────────────────────────────────────────────
+
+class CommandRequest(BaseModel):
+    command: str
+
+
+# ── Responses ───────────────────────────────────────────────────────────
+
+class CommandResponse(BaseModel):
+    status: str                       # success | error | unresolved
+    message: str
+    diff: Optional[dict[str, Any]] = None
+    suggestions: Optional[list[str]] = None
+
+
+class SessionInfo(BaseModel):
+    session_id: str
+    file_name: str
+    file_size_bytes: int
+    columns: list[dict[str, str]]     # [{"name": "salary", "dtype": "Float64"}]
+    row_count: int
+    status: str                       # active | processing
+
+
+class UploadResponse(BaseModel):
+    session_id: str
+    file_name: str
+    rows: int
+    columns: list[str]
+    size_human: str
+
+
+class ErrorResponse(BaseModel):
+    detail: str

requirements.txt

@@ -0,0 +1,12 @@
+fastapi==0.115.0
+uvicorn[standard]==0.30.0
+python-multipart==0.0.9
+polars==1.0.0
+rapidfuzz==3.9.0
+pydantic==2.9.0
+websockets==12.0
+aiosqlite==0.20.0
+httpx==0.27.0
+onnxruntime>=1.16.0
+numpy>=1.24.0
+llama-cpp-python>=0.2.85

services/__init__.py

@@ -0,0 +1 @@
+# services package

services/audit_service.py

@@ -0,0 +1,83 @@
+"""
+Audit service — SQLite-backed command log.
+
+Every executed command is recorded with enough info to reconstruct
+the inverse operation (for undo).  Uses aiosqlite so it doesn't
+block the async event loop.
+"""
+from __future__ import annotations
+
+import json
+import aiosqlite
+
+from config import AUDIT_DB_PATH
+
+_CREATE_TABLE = """
+CREATE TABLE IF NOT EXISTS audit_log (
+    id          INTEGER PRIMARY KEY AUTOINCREMENT,
+    session_id  TEXT    NOT NULL,
+    command     TEXT    NOT NULL,
+    intent      TEXT    NOT NULL,
+    diff        TEXT,
+    created_at  TEXT    DEFAULT (datetime('now'))
+);
+"""
+
+_db: aiosqlite.Connection | None = None
+
+
+async def _get_db() -> aiosqlite.Connection:
+    global _db
+    if _db is None:
+        _db = await aiosqlite.connect(AUDIT_DB_PATH)
+        await _db.execute(_CREATE_TABLE)
+        await _db.commit()
+    return _db
+
+
+async def log_command(
+    session_id: str,
+    command: str,
+    intent: dict | None,
+    diff: dict | None,
+) -> None:
+    """Insert an audit row."""
+    db = await _get_db()
+    await db.execute(
+        "INSERT INTO audit_log (session_id, command, intent, diff) VALUES (?, ?, ?, ?)",
+        (
+            session_id,
+            command,
+            json.dumps(intent or {}, ensure_ascii=False),
+            json.dumps(diff or {}, ensure_ascii=False),
+        ),
+    )
+    await db.commit()
+
+
+async def get_history(session_id: str, limit: int = 50) -> list[dict]:
+    """Fetch recent commands for a session (for undo UI)."""
+    db = await _get_db()
+    cursor = await db.execute(
+        "SELECT id, command, intent, diff, created_at FROM audit_log "
+        "WHERE session_id = ? ORDER BY id DESC LIMIT ?",
+        (session_id, limit),
+    )
+    rows = await cursor.fetchall()
+    return [
+        {
+            "id": r[0],
+            "command": r[1],
+            "intent": json.loads(r[2]),
+            "diff": json.loads(r[3]),
+            "created_at": r[4],
+        }
+        for r in rows
+    ]
+
+
+async def close() -> None:
+    global _db
+    if _db is not None:
+        await _db.close()
+        _db = None

services/file_manager.py

@@ -0,0 +1,137 @@
+"""
+File manager — upload, CSV→Parquet conversion, download, cleanup.
+"""
+from __future__ import annotations
+
+import os
+import uuid
+import shutil
+import polars as pl
+
+from config import DATA_DIR, UPLOAD_DIR, LAZY_THRESHOLD_BYTES, MAX_UPLOAD_BYTES
+from services.session_manager import session_manager
+from core.column_registry import column_registry
+
+
+def _human_size(n: int) -> str:
+    for unit in ("B", "KB", "MB", "GB"):
+        if n < 1024:
+            return f"{n:.1f} {unit}"
+        n /= 1024
+    return f"{n:.1f} TB"
+
+
+def _detect_separator(path: str) -> str:
+    """Sniff the separator from the first few KB."""
+    with open(path, "r", errors="replace") as f:
+        head = f.read(8192)
+    for sep in (",", ";", "\t", "|"):
+        if sep in head:
+            return sep
+    return ","
+
+
+def handle_upload(file_content: bytes, file_name: str) -> dict:
+    """Save uploaded file, convert to Parquet, create session."""
+    # Size check
+    if len(file_content) > MAX_UPLOAD_BYTES:
+        raise ValueError(f"File too large. Max {MAX_UPLOAD_BYTES // (1024**2)} MB allowed.")
+
+    # Generate session
+    session_id = uuid.uuid4().hex[:12]
+
+    # Save raw upload
+    ext = os.path.splitext(file_name)[1].lower()
+    raw_path = os.path.join(UPLOAD_DIR, f"{session_id}{ext}")
+    with open(raw_path, "wb") as f:
+        f.write(file_content)
+
+    # Determine file type and read
+    if ext in (".csv", ".tsv", ".txt"):
+        sep = _detect_separator(raw_path)
+        lf = pl.scan_csv(raw_path, separator=sep, try_parse_dates=True)
+    elif ext in (".xlsx", ".xls"):
+        # Polars can read Excel but needs the feature flag.
+        # Fall back to eager read for Excel.
+        df = pl.read_excel(raw_path)  # type: ignore[attr-defined]
+        parquet_path = os.path.join(DATA_DIR, f"{session_id}.parquet")
+        df.write_parquet(parquet_path)
+        os.remove(raw_path)
+    elif ext == ".parquet":
+        shutil.copy2(raw_path, os.path.join(DATA_DIR, f"{session_id}.parquet"))
+        os.remove(raw_path)
+        lf = pl.scan_parquet(os.path.join(DATA_DIR, f"{session_id}.parquet"))
+    else:
+        os.remove(raw_path)
+        raise ValueError(f"Unsupported file format: {ext}")
+
+    # For CSV: stream-convert to Parquet
+    if ext in (".csv", ".tsv", ".txt"):
+        parquet_path = os.path.join(DATA_DIR, f"{session_id}.parquet")
+        lf.sink_parquet(parquet_path)
+        os.remove(raw_path)
+        lf = pl.scan_parquet(parquet_path)
+
+    # Read metadata (no full data load)
+    schema = lf.collect_schema()
+    row_count = lf.select(pl.len()).collect().item()
+    columns = [{"name": name, "dtype": str(dtype)} for name, dtype in schema.items()]
+
+    # Register in session manager
+    session_manager.create(
+        session_id=session_id,
+        file_name=file_name,
+        file_size_bytes=len(file_content),
+        columns=columns,
+        row_count=row_count,
+    )
+
+    # Register columns for fuzzy resolution
+    column_registry.register(session_id, schema.names())
+
+    return {
+        "session_id": session_id,
+        "file_name": file_name,
+        "rows": row_count,
+        "columns": schema.names(),
+        "size_human": _human_size(len(file_content)),
+    }
+
+
+def get_download_path(session_id: str) -> Optional[str]:
+    """Return path to the Parquet file, or None if session doesn't exist."""
+    path = session_manager.get_filepath(session_id)
+    if os.path.exists(path) and session_manager.get(session_id):
+        return path
+    return None
+
+
+def export_to_csv(session_id: str) -> Optional[str]:
+    """Export Parquet to a temporary CSV file and return its path."""
+    pq_path = session_manager.get_filepath(session_id)
+    if not os.path.exists(pq_path):
+        return None
+
+    csv_path = os.path.join(UPLOAD_DIR, f"{session_id}.csv")
+    lf = pl.scan_parquet(pq_path)
+    if os.path.getsize(pq_path) > LAZY_THRESHOLD_BYTES:
+        lf.sink_csv(csv_path)
+    else:
+        lf.collect().write_csv(csv_path)
+    return csv_path
+
+
+def delete_session(session_id: str) -> bool:
+    """Remove session metadata, Parquet file, and column registry."""
+    import glob
+    pattern = os.path.join(DATA_DIR, f"{session_id}*.parquet")
+    removed = False
+    for pq_path in glob.glob(pattern):
+        try:
+            os.remove(pq_path)
+            removed = True
+        except Exception:
+            pass
+    session_manager.remove(session_id)
+    column_registry.remove(session_id)
+    return removed

services/session_manager.py

@@ -0,0 +1,120 @@
+"""
+Session manager — lightweight in-memory session metadata.
+
+In production, swap this dict for Redis.  Each entry holds only
+metadata (not the dataframe itself).  The actual data lives on
+disk as a Parquet file under DATA_DIR/{session_id}.parquet.
+"""
+from __future__ import annotations
+
+import time
+import threading
+from dataclasses import dataclass, field
+from typing import Optional
+
+from config import DATA_DIR, SESSION_TTL_MINUTES
+
+
+@dataclass
+class SessionMeta:
+    session_id: str
+    file_name: str
+    file_size_bytes: int
+    columns: list[dict[str, str]]   # [{"name": ..., "dtype": ...}]
+    row_count: int
+    status: str = "active"
+    current_version: int = 0
+    created_at: float = field(default_factory=time.time)
+    last_active: float = field(default_factory=time.time)
+
+    def touch(self) -> None:
+        self.last_active = time.time()
+
+
+class SessionManager:
+    """Thread-safe session store."""
+
+    def __init__(self) -> None:
+        self._sessions: dict[str, SessionMeta] = {}
+        self._lock = threading.Lock()
+
+    def create(
+        self,
+        session_id: str,
+        file_name: str,
+        file_size_bytes: int,
+        columns: list[dict[str, str]],
+        row_count: int,
+    ) -> SessionMeta:
+        meta = SessionMeta(
+            session_id=session_id,
+            file_name=file_name,
+            file_size_bytes=file_size_bytes,
+            columns=columns,
+            row_count=row_count,
+        )
+        with self._lock:
+            self._sessions[session_id] = meta
+        return meta
+
+    def get(self, session_id: str) -> Optional[SessionMeta]:
+        with self._lock:
+            return self._sessions.get(session_id)
+
+    def get_filepath(self, session_id: str) -> str:
+        """Get the absolute filepath of the current version of the Parquet file."""
+        meta = self.get(session_id)
+        if meta:
+            version = getattr(meta, "current_version", 0)
+            if version > 0:
+                v_path = os.path.join(DATA_DIR, f"{session_id}_v{version}.parquet")
+                if os.path.exists(v_path):
+                    return v_path
+            # Version 0 or fallback: check v0 path first
+            v0_path = os.path.join(DATA_DIR, f"{session_id}_v0.parquet")
+            if os.path.exists(v0_path):
+                return v0_path
+        return os.path.join(DATA_DIR, f"{session_id}.parquet")
+
+    def touch(self, session_id: str) -> None:
+        meta = self.get(session_id)
+        if meta:
+            meta.touch()
+
+    def remove(self, session_id: str) -> None:
+        with self._lock:
+            self._sessions.pop(session_id, None)
+
+    def list_active(self) -> list[SessionMeta]:
+        """Return sessions that haven't expired."""
+        now = time.time()
+        cutoff = now - SESSION_TTL_MINUTES * 60
+        with self._lock:
+            return [
+                m for m in self._sessions.values()
+                if m.last_active > cutoff
+            ]
+
+    def cleanup_expired(self) -> int:
+        """Remove expired sessions and their Parquet files. Returns count removed."""
+        import os
+        import glob
+        now = time.time()
+        cutoff = now - SESSION_TTL_MINUTES * 60
+        removed = 0
+        with self._lock:
+            expired = [sid for sid, m in self._sessions.items() if m.last_active <= cutoff]
+            for sid in expired:
+                del self._sessions[sid]
+                pattern = os.path.join(DATA_DIR, f"{sid}*.parquet")
+                for pq in glob.glob(pattern):
+                    try:
+                        os.remove(pq)
+                    except Exception:
+                        pass
+                removed += 1
+        return removed
+
+
+# Module-level singleton
+session_manager = SessionManager()

test_local_parser.py

@@ -0,0 +1,154 @@
+"""
+Unit tests for zero-llm-engine local parser components.
+"""
+from __future__ import annotations
+
+import os
+import sys
+
+# Insert current dir to path for imports
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+
+from core.parser.local_parser.spelling import SymSpell
+from core.parser.local_parser.synonyms import SynonymMapper
+from core.parser.local_parser.ast_extractor import SafeMathEvaluator
+from core.parser.local_parser.embeddings import WordPieceTokenizer, EmbeddingModel, cosine_similarity
+from core.parser.local_parser.parser import LocalIntentParser
+from core.column_registry import column_registry
+from services.file_manager import handle_upload, delete_session
+
+def test_spelling():
+    print("Testing spelling correction...")
+    sym = SymSpell(max_edit_distance=2)
+    sym.add_word("salary")
+    sym.add_word("average")
+    sym.add_word("decrease")
+    
+    # Simple typo correction
+    assert sym.lookup("salry")[0] == "salary"
+    assert sym.lookup("avrage")[0] == "average"
+    assert sym.lookup("decres")[0] == "decrease"
+    
+    # Sentence correction
+    corrected = sym.correct_query("decres the salry")
+    assert "decrease" in corrected
+    assert "salary" in corrected
+    print("✅ Spelling tests passed!")
+
+def test_synonyms():
+    print("Testing synonyms and Hinglish normalization...")
+    mapper = SynonymMapper()
+    
+    # Hinglish mapping
+    res = mapper.normalize_text("vetan ko 10% badhao")
+    assert "salary" in res
+    assert "increase" in res
+    
+    res2 = mapper.normalize_text("sabse chhota umar")
+    assert "min" in res2
+    assert "age" in res2
+    print("✅ Synonym tests passed!")
+
+def test_ast():
+    print("Testing safe AST evaluator...")
+    evaluator = SafeMathEvaluator(variables={"x": 10, "y": 20})
+    
+    # Math operations
+    assert evaluator.evaluate("x + y * 2") == 50
+    assert evaluator.evaluate("(y - x) / 2") == 5.0
+    
+    # Comparison and logic
+    assert evaluator.evaluate("x > 5 and y < 30") is True
+    assert evaluator.evaluate("x == 10 or y == 5") is True
+    
+    # Name error / safety check
+    try:
+        evaluator.evaluate("import os")
+        assert False, "Should raise exception for imports"
+    except Exception:
+        pass
+        
+    try:
+        evaluator.evaluate("x + z")
+        assert False, "Should raise exception for undefined variable"
+    except Exception:
+        pass
+    print("✅ AST tests passed!")
+
+def test_tokenizer_and_embeddings():
+    print("Testing tokenizer and embeddings...")
+    # Initialize embedding model (downloads if necessary)
+    model = EmbeddingModel()
+    model.load_model()
+    
+    # WordPiece encoding tests
+    tokenizer = model.tokenizer
+    encoded = tokenizer.encode("salary")
+    assert "input_ids" in encoded
+    assert encoded["input_ids"].shape == (1, 128)
+    
+    # Embedding generation
+    emb1 = model.get_embedding("salary")
+    emb2 = model.get_embedding("vetan")
+    emb3 = model.get_embedding("country")
+    
+    # L2 Normalized shape check
+    assert len(emb1.shape) == 1
+    assert emb1.shape[0] > 0
+    
+    # Cosine similarities
+    sim_salary = cosine_similarity(emb1, emb2)
+    sim_diff = cosine_similarity(emb1, emb3)
+    
+    print(f"Similarity (salary, vetan): {sim_salary}")
+    print(f"Similarity (salary, country): {sim_diff}")
+    
+    # Match column
+    cols = ["Age", "Salary", "Name", "Country"]
+    best_col, score = model.match_column("payrate", cols, threshold=0.3)
+    assert best_col == "Salary"
+    print("✅ Tokenizer and Embedding tests passed!")
+
+def test_orchestration():
+    print("Testing orchestration parser...")
+    # Register columns for test session
+    sid = "test_parser_session"
+    column_registry.register_columns(sid, [
+        {"name": "Name", "dtype": "String"},
+        {"name": "City", "dtype": "String"},
+        {"name": "Salary", "dtype": "Float64"},
+        {"name": "Age", "dtype": "Int64"},
+    ])
+    
+    parser = LocalIntentParser()
+    
+    # 1. Test increase
+    res1 = parser.parse_intent(sid, "salry ko 10% badhao")
+    assert res1 is not None
+    assert res1["operation"] == "increase"
+    assert res1["column"] == "Salary"
+    assert res1["value"] == 10.0
+    assert res1["is_percent"] is True
+    
+    # 2. Test filter
+    res2 = parser.parse_intent(sid, "salary se zyada 50000 dikhao")
+    # "se zyada" gets normalized, condition becomes >
+    assert res2 is not None
+    assert res2["operation"] == "filter"
+    assert res2["column"] == "Salary"
+    assert res2["condition"] in [">", "=="] # depends on normalization
+    
+    # Clean up
+    column_registry.clear_session(sid)
+    print("✅ Orchestration tests passed!")
+
+if __name__ == "__main__":
+    test_spelling()
+    test_synonyms()
+    test_ast()
+    try:
+        test_tokenizer_and_embeddings()
+    except Exception as e:
+        print(f"⚠️ Embeddings/Tokenizer test skipped or failed due to env: {e}")
+    test_orchestration()
+    print("\n🎉 ALL TESTS COMPLETED SUCCESSFULY!")

test_smoke.py

@@ -0,0 +1,138 @@
+"""
+End-to-end smoke test for the Zero-LLM engine.
+Creates a sample CSV, uploads it, runs commands, verifies results.
+"""
+import sys
+import os
+
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+
+import polars as pl
+from services.file_manager import handle_upload, delete_session
+from services.session_manager import session_manager
+from core.column_registry import column_registry
+from core.intent_parser import parse_intent
+from core.validator import validate_intent
+from core.router import dispatch
+
+
+def create_test_csv():
+    df = pl.DataFrame({
+        "Name": ["Rahul", "Priya", "Amit", "Sneha", "Vikram", "Neha", "Ravi", "Pooja"],
+        "City": ["Mumbai", "Delhi", "Mumbai", "Bangalore", "Delhi", "Mumbai", "Bangalore", "Delhi"],
+        "Salary": [50000, 60000, 55000, 70000, 65000, 48000, 72000, 58000],
+        "Age": [28, 32, 25, 35, 30, 27, 40, 29],
+        "Department": ["Engineering", "Marketing", "Engineering", "Design", "Marketing", "Engineering", "Design", "Marketing"],
+    })
+    base_dir = os.path.dirname(os.path.abspath(__file__))
+    uploads_dir = os.path.join(base_dir, "data", "uploads")
+    os.makedirs(uploads_dir, exist_ok=True)
+    path = os.path.join(uploads_dir, "test_sample.csv")
+    df.write_csv(path)
+    return path
+
+
+def run_tests():
+    print("=" * 60)
+    print("ZERO-LLM ENGINE — SMOKE TEST")
+    print("=" * 60)
+
+    # ── 1. Upload ───────────────────────────────────────────────
+    print("\n[1] Creating + uploading test CSV...")
+    csv_path = create_test_csv()
+    with open(csv_path, "rb") as f:
+        result = handle_upload(f.read(), "test_sample.csv")
+    sid = result["session_id"]
+    print(f"    Session: {sid}")
+    print(f"    Rows: {result['rows']}, Columns: {result['columns']}")
+
+    # ── 2. Intent Parsing Tests ─────────────────────────────────
+    test_commands = [
+        # (command, expected_operation, expected_column_or_None)
+        ("salary ko 10% badhao",       "increase",           "Salary"),
+        ("salary ghatao 15%",          "decrease",           "Salary"),
+        ("salary 50000 se zyada dikhao","filter",             "Salary"),
+        ("name sort karo chota se bada","sort_asc",           "Name"),
+        ("total salary batao",         "sum",                "Salary"),
+        ("average age nikalo",         "average",            "Age"),
+        ("duplicate hatao",            "remove_duplicates",  None),
+        ("age column hatao",           "delete_column",      "Age"),
+    ]
+
+    print("\n[2] Intent parsing tests:")
+    passed = 0
+    for cmd, expected_op, expected_col in test_commands:
+        intent = parse_intent(sid, cmd)
+        op = intent["operation"] if intent else "NONE"
+        col = intent.get("column") if intent else None
+        op_ok = op == expected_op
+        col_ok = (expected_col is None and col is None) or col == expected_col
+        status = "✅" if op_ok and col_ok else "❌"
+        if op_ok and col_ok:
+            passed += 1
+        print(f"    {status} '{cmd}' → op={op}, col={col}")
+
+    print(f"    Parsing: {passed}/{len(test_commands)} passed")
+
+    # ── 3. Execution Tests ──────────────────────────────────────
+    exec_commands = [
+        "salary ko 10% badhao",
+        "total salary batao",
+        "average salary nikalo",
+        "name sort karo chota se bada",
+        "sabse bada salary batao",
+        "sabse chhota salary batao",
+    ]
+
+    print("\n[3] Execution tests:")
+    for cmd in exec_commands:
+        intent = parse_intent(sid, cmd)
+        if intent is None:
+            print(f"    ❌ '{cmd}' → unresolved")
+            continue
+        error = validate_intent(sid, intent)
+        if error:
+            print(f"    ❌ '{cmd}' → validation: {error}")
+            continue
+        try:
+            result = dispatch(sid, intent)
+            print(f"    ✅ '{cmd}' → {result['message']}")
+        except Exception as e:
+            print(f"    ❌ '{cmd}' → error: {e}")
+
+    # ── 4. Filter test ──────────────────────────────────────────
+    print("\n[4] Filter test:")
+    intent = parse_intent(sid, "city Mumbai dikhao")
+    if intent:
+        error = validate_intent(sid, intent)
+        if error:
+            print(f"    ❌ Validation: {error}")
+        else:
+            result = dispatch(sid, intent)
+            print(f"    ✅ {result['message']}")
+
+    # ── 5. Verify final state ───────────────────────────────────
+    print("\n[5] Final data state:")
+    pq_path = f"/home/z/my-project/zero-llm-engine/data/sessions/{sid}.parquet"
+    if os.path.exists(pq_path):
+        df = pl.read_parquet(pq_path)
+        print(f"    Rows: {len(df)}, Columns: {df.columns}")
+        print(df)
+
+    # ── Cleanup ─────────────────────────────────────────────────
+    delete_session(sid)
+    if os.path.exists(csv_path):
+        os.remove(csv_path)
+    print("\n[6] Cleanup done.")
+
+    # ── Server start instructions ───────────────────────────────
+    print("\n" + "=" * 60)
+    print("To start the server:")
+    print("  cd /home/z/my-project/zero-llm-engine")
+    print("  source venv/bin/activate")
+    print("  uvicorn main:app --host 0.0.0.0 --port 8000 --reload")
+    print("=" * 60)
+
+
+if __name__ == "__main__":
+    run_tests()