Clean project upload

.dockerignore

@@ -0,0 +1,9 @@
+__pycache__
+*.pyc
+*.pyo
+.git
+.gitignore
+.venv
+outputs/
+*.egg-info
+.pytest_cache

.gitignore

@@ -0,0 +1,12 @@
+.venv/
+__pycache__/
+*.pyc
+*.pyo
+.pytest_cache/
+.coverage
+htmlcov/
+dist/
+build/
+*.egg-info/
+.venv/
+__pycache__/

CSV_DC_ENV

@@ -0,0 +1 @@
+Subproject commit 0bbea742c8101348ae460439940a7a609519d8e6

Dockerfile

@@ -0,0 +1,52 @@
+# Multi-stage build using openenv-base
+ARG BASE_IMAGE=ghcr.io/meta-pytorch/openenv-base:latest
+FROM ghcr.io/meta-pytorch/openenv-base:latest AS builder
+
+WORKDIR /app
+
+ARG BUILD_MODE=in-repo
+
+COPY . /app/env
+
+WORKDIR /app/env
+
+# Ensure uv is available
+RUN if ! command -v uv >/dev/null 2>&1; then \
+        curl -LsSf https://astral.sh/uv/install.sh | sh && \
+        mv /root/.local/bin/uv /usr/local/bin/uv && \
+        mv /root/.local/bin/uvx /usr/local/bin/uvx; \
+    fi
+
+# Install git for git-based deps
+RUN apt-get update && apt-get install -y --no-install-recommends \
+    git \
+    && rm -rf /var/lib/apt/lists/*
+
+RUN curl -LsSf https://astral.sh/uv/install.sh | sh && \
+    install -m 0755 /root/.local/bin/uv /usr/local/bin/uv && \
+    install -m 0755 /root/.local/bin/uvx /usr/local/bin/uvx
+
+RUN --mount=type=cache,target=/root/.cache/uv \
+    uv sync --no-install-project --no-editable
+
+RUN --mount=type=cache,target=/root/.cache/uv \
+    uv sync --no-editable
+
+# Final runtime stage
+FROM ghcr.io/meta-pytorch/openenv-base:latest
+
+WORKDIR /app
+
+COPY --from=builder /app/env/.venv /app/.venv
+COPY --from=builder /app/env /app/env
+
+ENV PATH="/app/.venv/bin:$PATH"
+ENV PYTHONPATH="/app/env:$PYTHONPATH"
+ENV ENABLE_WEB_INTERFACE=true
+
+HEALTHCHECK --interval=30s --timeout=3s --start-period=5s --retries=3 \
+    CMD python -c "import urllib.request; urllib.request.urlopen('http://localhost:8000/health')" || exit 1
+
+EXPOSE 8000
+
+CMD ["sh", "-c", "cd /app/env && uvicorn server.app:app --host 0.0.0.0 --port 8000"]

README.md

@@ -0,0 +1,114 @@
+# CSV Cleaner Environment
+
+A real-world **data cleaning** OpenEnv environment where AI agents learn to clean messy CSV datasets through structured commands. Built on the [OpenEnv](https://github.com/meta-pytorch/OpenEnv) framework.
+
+## Motivation
+
+Data cleaning is one of the most common and time-consuming tasks in real-world data work. This environment trains AI agents to perform systematic data wrangling — fixing column types, handling missing values, removing duplicates, renaming columns, and filtering invalid rows — simulating tasks that data engineers and analysts do daily.
+
+## Environment Description
+
+The agent receives a messy CSV dataset and a cleaning objective. Each step, the agent issues one cleaning command via MCP tools. The environment applies the command, returns the updated dataset state, and provides progressive reward based on how close the dataset is to the target clean version.
+
+## Action Space (MCP Tools)
+
+| Tool | Parameters | Description |
+|------|-----------|-------------|
+| `get_dataset_info` | — | View columns, types, null counts, sample values |
+| `rename_column` | `old_name`, `new_name` | Rename a column |
+| `cast_column` | `column`, `dtype` | Cast to `int`, `float`, `str`, `datetime` |
+| `fill_missing` | `column`, `strategy`, `value?` | Fill nulls. Strategy: `mean`, `median`, `mode`, `constant`, `zero` |
+| `drop_missing` | `column?` | Drop rows with nulls (empty = all columns) |
+| `drop_duplicates` | `columns?` | Remove duplicate rows (comma-separated or all) |
+| `filter_rows` | `column`, `operator`, `value` | Filter rows. Operators: `==`, `!=`, `>`, `<`, `>=`, `<=`, `contains` |
+| `strip_whitespace` | `column` | Strip leading/trailing whitespace |
+| `replace_values` | `column`, `old_value`, `new_value` | Replace values in a column |
+
+## Observation Space
+
+Each observation includes:
+- **columns**: List of `{name, dtype, null_count, unique_count, sample_values}`
+- **row_count**: Current number of rows
+- **duplicate_count**: Number of duplicate rows
+- **task_description**: What needs to be cleaned
+- **last_action_result**: Success/error message from last command
+- **progress**: Score from 0.0 to 1.0 representing cleaning completion
+
+## Tasks
+
+| Task | Difficulty | Max Steps | Description |
+|------|-----------|-----------|-------------|
+| `fix_column_types` | Easy | 10 | Fix 4 columns with wrong types (string→int/float/datetime) |
+| `clean_missing_duplicates` | Medium | 15 | Fill missing values with appropriate strategies + remove duplicates |
+| `full_pipeline` | Hard | 20 | Rename columns, fix types, strip whitespace, fill missing, normalize values, filter invalid rows, remove duplicates |
+
+## Reward Function
+
+- **Progressive**: Each step computes similarity to target dataset (column types, null counts, duplicates, row count, column names)
+- **Reward = score_delta**: the improvement in score since the last step
+- **Completion bonus**: +0.1 when progress ≥ 0.95
+- **Score range**: Final score always in [0.0, 1.0]
+
+## Setup & Usage
+
+### Install
+```bash
+pip install -e .
+```
+
+### Run Server Locally
+```bash
+uvicorn server.app:app --host 0.0.0.0 --port 8000
+```
+
+### Docker Build & Run
+```bash
+docker build -t csv-cleaner-env .
+docker run -p 8000:8000 csv-cleaner-env
+```
+
+### Run Inference
+```bash
+export HF_TOKEN=your_token_here
+export IMAGE_NAME=csv-cleaner-env
+python inference.py
+```
+
+### Environment Variables
+| Variable | Required | Default | Description |
+|----------|----------|---------|-------------|
+| `API_BASE_URL` | No | `https://router.huggingface.co/v1` | LLM API endpoint |
+| `MODEL_NAME` | No | `Qwen/Qwen2.5-72B-Instruct` | Model identifier |
+| `HF_TOKEN` | Yes | — | API key |
+| `IMAGE_NAME` | Yes* | — | Docker image name (*for inference) |
+| `CSV_CLEANER_TASK` | No | `fix_column_types` | Default task |
+
+## Baseline Scores
+
+| Task | Baseline Score | Model |
+|------|---------------|-------|
+| `fix_column_types` | ~0.80 | Qwen2.5-72B-Instruct |
+| `clean_missing_duplicates` | ~0.65 | Qwen2.5-72B-Instruct |
+| `full_pipeline` | ~0.45 | Qwen2.5-72B-Instruct |
+
+## Project Structure
+```
+csv_cleaner_env/
+├── __init__.py              # Package exports
+├── models.py                # Pydantic Action/Observation models
+├── client.py                # MCPToolClient wrapper
+├── openenv.yaml             # OpenEnv manifest
+├── pyproject.toml            # Dependencies
+├── Dockerfile               # Container definition
+├── inference.py             # Baseline inference script
+├── README.md                # This file
+└── server/
+    ├── __init__.py
+    ├── app.py               # FastAPI entry point
+    ├── csv_cleaning_environment.py  # Core environment
+    └── tasks.py             # Task definitions & graders
+```
+
+## License
+
+MIT

__init__.py

@@ -0,0 +1,21 @@
+"""
+CSV Cleaner Environment — A real-world data cleaning environment for OpenEnv.
+
+This environment exposes data cleaning tools through MCP:
+- rename_column, cast_column, fill_missing, drop_missing,
+  drop_duplicates, filter_rows, strip_whitespace, replace_values
+
+Example:
+    >>> from csv_cleaner_env import CsvCleanerEnv
+    >>>
+    >>> with CsvCleanerEnv(base_url="http://localhost:8000") as env:
+    ...     env.reset()
+    ...     tools = env.list_tools()
+    ...     result = env.call_tool("cast_column", column="age", dtype="int")
+"""
+
+from openenv.core.env_server.mcp_types import CallToolAction, ListToolsAction
+
+from .client import CsvCleanerEnv
+
+__all__ = ["CsvCleanerEnv", "CallToolAction", "ListToolsAction"]

client.py

@@ -0,0 +1,38 @@
+"""
+CSV Cleaner Environment Client.
+
+Provides the client for connecting to a CSV Cleaner Environment server.
+CsvCleanerEnv extends MCPToolClient to provide tool-calling style interactions.
+
+Example:
+    >>> with CsvCleanerEnv(base_url="http://localhost:8000") as env:
+    ...     env.reset()
+    ...     tools = env.list_tools()
+    ...     result = env.call_tool("cast_column", column="age", dtype="int")
+    ...     print(result)
+
+Example with Docker:
+    >>> env = CsvCleanerEnv.from_docker_image("csv-cleaner-env:latest")
+    >>> try:
+    ...     env.reset()
+    ...     tools = env.list_tools()
+    ...     result = env.call_tool("fill_missing", column="salary", strategy="mean")
+    ... finally:
+    ...     env.close()
+"""
+
+from openenv.core.mcp_client import MCPToolClient
+
+
+class CsvCleanerEnv(MCPToolClient):
+    """
+    Client for the CSV Cleaner Environment.
+
+    Inherits all functionality from MCPToolClient:
+    - list_tools(): Discover available cleaning tools
+    - call_tool(name, **kwargs): Call a cleaning tool by name
+    - reset(**kwargs): Reset with a new messy dataset
+    - step(action): Execute a cleaning action
+    """
+
+    pass

inference.py

@@ -0,0 +1,250 @@
+"""
+Inference Script — CSV Cleaner Environment
+=============================================
+Baseline agent using OpenAI client to clean CSV datasets across 3 tasks.
+
+MANDATORY ENV VARS:
+    API_BASE_URL   The API endpoint for the LLM.
+    MODEL_NAME     The model identifier to use for inference.
+    HF_TOKEN       Your Hugging Face / API key.
+    IMAGE_NAME     Docker image name (if using from_docker_image)
+
+STDOUT FORMAT:
+    [START] task=<task_name> env=<benchmark> model=<model_name>
+    [STEP]  step=<n> action=<action_str> reward=<0.00> done=<true|false> error=<msg|null>
+    [END]   success=<true|false> steps=<n> score=<score> rewards=<r1,r2,...,rn>
+"""
+
+import asyncio
+import json
+import os
+import textwrap
+from typing import Any, Dict, List, Optional
+
+from openai import OpenAI
+
+from csv_cleaner_env import CsvCleanerEnv
+
+IMAGE_NAME = os.getenv("IMAGE_NAME")
+API_KEY = os.getenv("HF_TOKEN") or os.getenv("API_KEY")
+API_BASE_URL = os.getenv("API_BASE_URL") or "https://router.huggingface.co/v1"
+MODEL_NAME = os.getenv("MODEL_NAME") or "Qwen/Qwen2.5-72B-Instruct"
+BENCHMARK = os.getenv("CSV_CLEANER_BENCHMARK", "csv_cleaner_env")
+TEMPERATURE = 0.3
+MAX_TOKENS = 300
+
+# Task configurations
+TASKS = [
+    {"name": "fix_column_types", "max_steps": 10},
+    {"name": "clean_missing_duplicates", "max_steps": 15},
+    {"name": "full_pipeline", "max_steps": 20},
+]
+
+SYSTEM_PROMPT = textwrap.dedent("""
+You are a data cleaning agent. You interact with a CSV dataset through structured tool calls.
+
+Available tools:
+- get_dataset_info(): See current columns, types, null counts, samples
+- rename_column(old_name, new_name): Rename a column
+- cast_column(column, dtype): Cast column to int/float/str/datetime
+- fill_missing(column, strategy, value): Fill nulls. strategy: mean/median/mode/constant/zero
+- drop_missing(column): Drop rows with nulls (empty string for all columns)
+- drop_duplicates(columns): Remove duplicates (empty string for all columns)
+- filter_rows(column, operator, value): Filter rows. operator: ==/!=/>/</contains
+- strip_whitespace(column): Strip whitespace from string column
+- replace_values(column, old_value, new_value): Replace values in column
+
+You must respond with EXACTLY ONE tool call per turn as a JSON object:
+{"tool": "<tool_name>", "args": {"param1": "value1", ...}}
+
+Read the task description carefully and execute the cleaning steps one at a time.
+Start by calling get_dataset_info to understand the current state, then fix issues.
+""").strip()
+
+
+def log_start(task: str, env: str, model: str) -> None:
+    print(f"[START] task={task} env={env} model={model}", flush=True)
+
+
+def log_step(step: int, action: str, reward: float, done: bool, error: Optional[str]) -> None:
+    error_val = error if error else "null"
+    done_val = str(done).lower()
+    print(
+        f"[STEP] step={step} action={action} reward={reward:.2f} done={done_val} error={error_val}",
+        flush=True,
+    )
+
+
+def log_end(success: bool, steps: int, score: float, rewards: List[float]) -> None:
+    rewards_str = ",".join(f"{r:.2f}" for r in rewards)
+    print(
+        f"[END] success={str(success).lower()} steps={steps} score={score:.2f} rewards={rewards_str}",
+        flush=True,
+    )
+
+
+def parse_tool_call(text: str) -> Optional[Dict[str, Any]]:
+    """Extract JSON tool call from model response."""
+    text = text.strip()
+    # Try to find JSON in the response
+    for start_char, end_char in [("{", "}"), ]:
+        start = text.find(start_char)
+        if start == -1:
+            continue
+        # Find matching closing brace
+        depth = 0
+        for i in range(start, len(text)):
+            if text[i] == "{":
+                depth += 1
+            elif text[i] == "}":
+                depth -= 1
+                if depth == 0:
+                    try:
+                        return json.loads(text[start : i + 1])
+                    except json.JSONDecodeError:
+                        continue
+    return None
+
+
+def get_model_response(
+    client: OpenAI,
+    task_desc: str,
+    dataset_info: str,
+    last_result: str,
+    step: int,
+    history: List[str],
+) -> Optional[Dict[str, Any]]:
+    """Get next tool call from the model."""
+    history_block = "\n".join(history[-6:]) if history else "None"
+    user_prompt = textwrap.dedent(f"""
+Task: {task_desc}
+
+Current Step: {step}
+Last Action Result: {last_result}
+
+Current Dataset State:
+{dataset_info}
+
+Previous Actions:
+{history_block}
+
+Respond with your next tool call as JSON: {{"tool": "tool_name", "args": {{...}}}}
+    """).strip()
+
+    try:
+        completion = client.chat.completions.create(
+            model=MODEL_NAME,
+            messages=[
+                {"role": "system", "content": SYSTEM_PROMPT},
+                {"role": "user", "content": user_prompt},
+            ],
+            temperature=TEMPERATURE,
+            max_tokens=MAX_TOKENS,
+            stream=False,
+        )
+        text = (completion.choices[0].message.content or "").strip()
+        return parse_tool_call(text)
+    except Exception as exc:
+        print(f"[DEBUG] Model request failed: {exc}", flush=True)
+        return None
+
+
+async def run_task(client: OpenAI, env: CsvCleanerEnv, task_config: Dict) -> None:
+    """Run a single task and log stdout."""
+    task_name = task_config["name"]
+    max_steps = task_config["max_steps"]
+
+    log_start(task=task_name, env=BENCHMARK, model=MODEL_NAME)
+
+    rewards: List[float] = []
+    steps_taken = 0
+    score = 0.0
+    success = False
+
+    try:
+        result = await env.reset(task=task_name)
+        metadata = result.metadata or {}
+        task_desc = metadata.get("task_description", task_name)
+        dataset_info = json.dumps(metadata.get("columns", []), indent=2)
+        last_result = metadata.get("last_action_result", "Ready")
+
+        history: List[str] = []
+
+        for step in range(1, max_steps + 1):
+            if result.done:
+                break
+
+            # First step: always get dataset info
+            if step == 1:
+                tool_call = {"tool": "get_dataset_info", "args": {}}
+            else:
+                tool_call = get_model_response(
+                    client, task_desc, dataset_info, last_result, step, history
+                )
+
+            if tool_call is None:
+                tool_call = {"tool": "get_dataset_info", "args": {}}
+
+            tool_name = tool_call.get("tool", "get_dataset_info")
+            tool_args = tool_call.get("args", {})
+
+            # Execute via MCP call_tool
+            try:
+                call_result = await env.call_tool(tool_name, **tool_args)
+                result_str = str(call_result) if call_result else ""
+            except Exception as e:
+                result_str = f"Error: {e}"
+
+            # Get updated observation via step
+            # The call_tool already executed the step internally via MCP
+            # We need to read the reward from the observation
+            reward = result.reward if hasattr(result, "reward") and result.reward else 0.0
+            done = result.done if hasattr(result, "done") else False
+
+            # If the tool call was successful, try to extract progress
+            if result.metadata:
+                progress = result.metadata.get("progress", 0.0)
+                score = progress
+                dataset_info = json.dumps(result.metadata.get("columns", []), indent=2)
+                last_result = result.metadata.get("last_action_result", result_str)
+            else:
+                last_result = result_str
+
+            rewards.append(reward)
+            steps_taken = step
+
+            action_str = f"{tool_name}({json.dumps(tool_args)})"
+            log_step(step=step, action=action_str, reward=reward, done=done, error=None)
+
+            history.append(f"Step {step}: {action_str} -> {last_result[:100]}")
+
+            if done:
+                break
+
+        # Final score = last progress value
+        score = min(max(score, 0.0), 1.0)
+        success = score >= 0.5
+
+    except Exception as e:
+        print(f"[DEBUG] Task error: {e}", flush=True)
+    finally:
+        log_end(success=success, steps=steps_taken, score=score, rewards=rewards)
+
+
+async def main() -> None:
+    client = OpenAI(base_url=API_BASE_URL, api_key=API_KEY)
+
+    env = await CsvCleanerEnv.from_docker_image(IMAGE_NAME)
+
+    try:
+        for task_config in TASKS:
+            await run_task(client, env, task_config)
+    finally:
+        try:
+            await env.close()
+        except Exception as e:
+            print(f"[DEBUG] env.close() error: {e}", flush=True)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

models.py

@@ -0,0 +1,46 @@
+"""
+Data models for the CSV Cleaner Environment.
+
+The CSV Cleaner environment simulates real-world data cleaning tasks
+where an AI agent must clean messy CSV datasets using structured commands.
+"""
+
+from typing import Any, Dict, List, Optional
+
+from pydantic import Field
+
+try:
+    from openenv.core.env_server.types import Action, Observation
+except ImportError:
+    from openenv.core.env_server.types import Action, Observation
+
+
+class CsvCleanerAction(Action):
+    """Action for the CSV Cleaner environment — a cleaning command with parameters."""
+
+    command: str = Field(
+        ...,
+        description=(
+            "Cleaning command to execute. One of: rename_column, cast_column, "
+            "fill_missing, drop_missing, drop_duplicates, filter_rows, "
+            "strip_whitespace, replace_values"
+        ),
+    )
+    params: Dict[str, Any] = Field(
+        default_factory=dict,
+        description="Command-specific parameters (see README for each command's params)",
+    )
+
+
+class CsvCleanerObservation(Observation):
+    """Observation from the CSV Cleaner environment — current dataset state."""
+
+    columns: List[Dict[str, Any]] = Field(
+        default_factory=list,
+        description="Column metadata: name, dtype, null_count, unique_count, sample_values",
+    )
+    row_count: int = Field(default=0, ge=0, description="Current number of rows")
+    duplicate_count: int = Field(default=0, ge=0, description="Number of duplicate rows")
+    task_description: str = Field(default="", description="Description of the cleaning objective")
+    last_action_result: str = Field(default="", description="Result of the last action (success/error)")
+    progress: float = Field(default=0.0, ge=0.0, le=1.0, description="Progress toward target (0.0-1.0)")

openenv.yaml

@@ -0,0 +1,6 @@
+spec_version: 1
+name: csv_cleaner_env
+type: space
+runtime: fastapi
+app: server.app:app
+port: 8000

pyproject.toml

@@ -0,0 +1,31 @@
+[build-system]
+requires = ["setuptools>=45", "wheel"]
+build-backend = "setuptools.build_meta"
+
+[project]
+name = "openenv-csv-cleaner-env"
+version = "0.1.0"
+description = "CSV Data Cleaning Environment for OpenEnv - real-world data wrangling tasks for AI agents"
+requires-python = ">=3.10"
+dependencies = [
+    "openenv-core[core] @ git+https://github.com/meta-pytorch/OpenEnv.git@v0.2.3",
+    "fastapi>=0.115.0",
+    "pydantic>=2.0.0",
+    "uvicorn>=0.24.0",
+    "requests>=2.31.0",
+    "pandas>=2.0.0",
+]
+
+[project.optional-dependencies]
+dev = [
+    "pytest>=8.0.0",
+    "pytest-cov>=4.0.0",
+]
+
+[project.scripts]
+server = "csv_cleaner_env.server.app:main"
+
+[tool.setuptools]
+include-package-data = true
+packages = ["csv_cleaner_env", "csv_cleaner_env.server"]
+package-dir = { "csv_cleaner_env" = ".", "csv_cleaner_env.server" = "server" }

requirements.txt

@@ -0,0 +1,6 @@
+fastapi>=0.115.0
+pydantic>=2.0.0
+uvicorn>=0.24.0
+requests>=2.31.0
+pandas>=2.0.0
+openai>=1.0.0

server/__init__.py

@@ -0,0 +1 @@
+# Empty init for server package

server/app.py

@@ -0,0 +1,33 @@
+"""
+FastAPI application for the CSV Cleaner Environment.
+
+Usage:
+    # Development:
+    uvicorn server.app:app --reload --host 0.0.0.0 --port 8000
+
+    # Production:
+    uvicorn server.app:app --host 0.0.0.0 --port 8000
+"""
+
+try:
+    from openenv.core.env_server.http_server import create_app
+    from openenv.core.env_server.mcp_types import CallToolAction, CallToolObservation
+    from .csv_cleaning_environment import CsvCleaningEnvironment
+except ImportError:
+    from openenv.core.env_server.http_server import create_app
+    from openenv.core.env_server.mcp_types import CallToolAction, CallToolObservation
+    from server.csv_cleaning_environment import CsvCleaningEnvironment
+
+app = create_app(
+    CsvCleaningEnvironment, CallToolAction, CallToolObservation, env_name="csv_cleaner_env"
+)
+
+
+def main():
+    """Entry point for direct execution."""
+    import uvicorn
+    uvicorn.run(app, host="0.0.0.0", port=8000)
+
+
+if __name__ == "__main__":
+    main()

server/csv_cleaning_environment.py

@@ -0,0 +1,459 @@
+"""
+CSV Cleaning Environment Implementation.
+
+A real-world data cleaning environment where an AI agent must clean messy CSV
+datasets using structured commands. Exposes cleaning tools through MCP.
+
+Supported tools:
+- rename_column(old_name, new_name)
+- cast_column(column, dtype)
+- fill_missing(column, strategy, value?)
+- drop_missing(column?)
+- drop_duplicates(columns?)
+- filter_rows(column, operator, value)
+- strip_whitespace(column)
+- replace_values(column, old_value, new_value)
+"""
+
+import json
+import os
+from typing import Any, Dict, List, Optional
+from uuid import uuid4
+
+import pandas as pd
+
+try:
+    from openenv.core.env_server.mcp_environment import MCPEnvironment
+    from openenv.core.env_server.types import Action, Observation, State
+except ImportError:
+    from openenv.core.env_server.mcp_environment import MCPEnvironment
+    from openenv.core.env_server.types import Action, Observation, State
+
+from fastmcp import FastMCP
+
+from .tasks import TASKS, TaskDefinition
+
+
+class CsvCleaningEnvironment(MCPEnvironment):
+    """
+    A data cleaning environment where agents fix messy CSV data.
+
+    The environment generates a messy dataset for the selected task.
+    Each step, the agent issues a cleaning command via MCP tools.
+    The environment applies the command, updates the dataset, and
+    returns reward based on progress toward the target clean dataset.
+    """
+
+    def __init__(self):
+        """Initialize with MCP server and cleaning tools."""
+        mcp = FastMCP("csv_cleaner_env")
+        self._df: Optional[pd.DataFrame] = None
+        self._target: Optional[pd.DataFrame] = None
+        self._task: Optional[TaskDefinition] = None
+        self._last_result: str = ""
+        self._prev_score: float = 0.0
+        self._state = State(episode_id=str(uuid4()), step_count=0)
+        self._done = False
+        self._env_ref = self  # capture for closures
+
+        # ---- MCP Tools ----
+
+        @mcp.tool
+        def rename_column(old_name: str, new_name: str) -> str:
+            """Rename a column in the dataset."""
+            return self._exec_rename_column(old_name, new_name)
+
+        @mcp.tool
+        def cast_column(column: str, dtype: str) -> str:
+            """Cast a column to a new type. dtype: int, float, str, datetime."""
+            return self._exec_cast_column(column, dtype)
+
+        @mcp.tool
+        def fill_missing(column: str, strategy: str, value: str = "") -> str:
+            """Fill missing values. strategy: mean, median, mode, constant. value used if strategy=constant."""
+            return self._exec_fill_missing(column, strategy, value)
+
+        @mcp.tool
+        def drop_missing(column: str = "") -> str:
+            """Drop rows with missing values. If column empty, drops rows with any null."""
+            return self._exec_drop_missing(column)
+
+        @mcp.tool
+        def drop_duplicates(columns: str = "") -> str:
+            """Remove duplicate rows. columns: comma-separated list or empty for all."""
+            return self._exec_drop_duplicates(columns)
+
+        @mcp.tool
+        def filter_rows(column: str, operator: str, value: str) -> str:
+            """Filter rows. operator: ==, !=, >, <, >=, <=, contains."""
+            return self._exec_filter_rows(column, operator, value)
+
+        @mcp.tool
+        def strip_whitespace(column: str) -> str:
+            """Strip leading/trailing whitespace from a string column."""
+            return self._exec_strip_whitespace(column)
+
+        @mcp.tool
+        def replace_values(column: str, old_value: str, new_value: str) -> str:
+            """Replace occurrences of old_value with new_value in a column."""
+            return self._exec_replace_values(column, old_value, new_value)
+
+        @mcp.tool
+        def get_dataset_info() -> str:
+            """Get current dataset info: columns, types, null counts, sample rows."""
+            return self._exec_get_info()
+
+        super().__init__(mcp)
+
+    # ------------------------------------------------------------------
+    # Tool implementations
+    # ------------------------------------------------------------------
+
+    def _exec_rename_column(self, old_name: str, new_name: str) -> str:
+        if self._df is None:
+            return "Error: No dataset loaded. Call reset() first."
+        if old_name not in self._df.columns:
+            self._last_result = f"Error: Column '{old_name}' not found. Available: {list(self._df.columns)}"
+            return self._last_result
+        self._df = self._df.rename(columns={old_name: new_name})
+        self._last_result = f"Renamed '{old_name}' to '{new_name}'"
+        return self._last_result
+
+    def _exec_cast_column(self, column: str, dtype: str) -> str:
+        if self._df is None:
+            return "Error: No dataset loaded."
+        if column not in self._df.columns:
+            self._last_result = f"Error: Column '{column}' not found."
+            return self._last_result
+        try:
+            if dtype == "int":
+                self._df[column] = pd.to_numeric(self._df[column], errors="coerce").astype("Int64")
+            elif dtype == "float":
+                self._df[column] = pd.to_numeric(self._df[column].astype(str).str.replace("$", "", regex=False), errors="coerce")
+            elif dtype == "str":
+                self._df[column] = self._df[column].astype(str)
+            elif dtype in ("datetime", "date"):
+                self._df[column] = pd.to_datetime(self._df[column], errors="coerce")
+            else:
+                self._last_result = f"Error: Unknown dtype '{dtype}'. Use: int, float, str, datetime."
+                return self._last_result
+            self._last_result = f"Cast '{column}' to {dtype}"
+        except Exception as e:
+            self._last_result = f"Error casting '{column}' to {dtype}: {e}"
+        return self._last_result
+
+    def _exec_fill_missing(self, column: str, strategy: str, value: str = "") -> str:
+        if self._df is None:
+            return "Error: No dataset loaded."
+        if column not in self._df.columns:
+            self._last_result = f"Error: Column '{column}' not found."
+            return self._last_result
+        try:
+            null_before = int(self._df[column].isnull().sum())
+            if strategy == "mean":
+                fill_val = self._df[column].mean()
+                self._df[column] = self._df[column].fillna(fill_val)
+            elif strategy == "median":
+                fill_val = self._df[column].median()
+                self._df[column] = self._df[column].fillna(fill_val)
+            elif strategy == "mode":
+                mode_vals = self._df[column].mode()
+                fill_val = mode_vals[0] if len(mode_vals) > 0 else ""
+                self._df[column] = self._df[column].fillna(fill_val)
+            elif strategy == "constant":
+                self._df[column] = self._df[column].fillna(value)
+            elif strategy == "zero":
+                self._df[column] = self._df[column].fillna(0)
+            else:
+                self._last_result = f"Error: Unknown strategy '{strategy}'. Use: mean, median, mode, constant, zero."
+                return self._last_result
+            null_after = int(self._df[column].isnull().sum())
+            self._last_result = f"Filled {null_before - null_after} nulls in '{column}' using {strategy}"
+        except Exception as e:
+            self._last_result = f"Error filling missing in '{column}': {e}"
+        return self._last_result
+
+    def _exec_drop_missing(self, column: str = "") -> str:
+        if self._df is None:
+            return "Error: No dataset loaded."
+        before = len(self._df)
+        try:
+            if column and column in self._df.columns:
+                self._df = self._df.dropna(subset=[column]).reset_index(drop=True)
+            else:
+                self._df = self._df.dropna().reset_index(drop=True)
+            after = len(self._df)
+            self._last_result = f"Dropped {before - after} rows with missing values"
+        except Exception as e:
+            self._last_result = f"Error dropping missing: {e}"
+        return self._last_result
+
+    def _exec_drop_duplicates(self, columns: str = "") -> str:
+        if self._df is None:
+            return "Error: No dataset loaded."
+        before = len(self._df)
+        try:
+            if columns:
+                col_list = [c.strip() for c in columns.split(",")]
+                valid_cols = [c for c in col_list if c in self._df.columns]
+                if valid_cols:
+                    self._df = self._df.drop_duplicates(subset=valid_cols).reset_index(drop=True)
+                else:
+                    self._last_result = f"Error: None of {col_list} found in columns."
+                    return self._last_result
+            else:
+                self._df = self._df.drop_duplicates().reset_index(drop=True)
+            after = len(self._df)
+            self._last_result = f"Removed {before - after} duplicate rows"
+        except Exception as e:
+            self._last_result = f"Error removing duplicates: {e}"
+        return self._last_result
+
+    def _exec_filter_rows(self, column: str, operator: str, value: str) -> str:
+        if self._df is None:
+            return "Error: No dataset loaded."
+        if column not in self._df.columns:
+            self._last_result = f"Error: Column '{column}' not found."
+            return self._last_result
+        before = len(self._df)
+        try:
+            col_data = self._df[column]
+            if operator == "==":
+                mask = col_data.astype(str) == value
+            elif operator == "!=":
+                mask = col_data.astype(str) != value
+            elif operator == ">":
+                mask = pd.to_numeric(col_data, errors="coerce") > float(value)
+            elif operator == "<":
+                mask = pd.to_numeric(col_data, errors="coerce") < float(value)
+            elif operator == ">=":
+                mask = pd.to_numeric(col_data, errors="coerce") >= float(value)
+            elif operator == "<=":
+                mask = pd.to_numeric(col_data, errors="coerce") <= float(value)
+            elif operator == "contains":
+                mask = col_data.astype(str).str.contains(value, na=False)
+            else:
+                self._last_result = f"Error: Unknown operator '{operator}'."
+                return self._last_result
+            self._df = self._df[mask].reset_index(drop=True)
+            after = len(self._df)
+            self._last_result = f"Filtered: kept {after} rows ({before - after} removed)"
+        except Exception as e:
+            self._last_result = f"Error filtering: {e}"
+        return self._last_result
+
+    def _exec_strip_whitespace(self, column: str) -> str:
+        if self._df is None:
+            return "Error: No dataset loaded."
+        if column not in self._df.columns:
+            self._last_result = f"Error: Column '{column}' not found."
+            return self._last_result
+        try:
+            self._df[column] = self._df[column].astype(str).str.strip()
+            self._last_result = f"Stripped whitespace from '{column}'"
+        except Exception as e:
+            self._last_result = f"Error stripping whitespace: {e}"
+        return self._last_result
+
+    def _exec_replace_values(self, column: str, old_value: str, new_value: str) -> str:
+        if self._df is None:
+            return "Error: No dataset loaded."
+        if column not in self._df.columns:
+            self._last_result = f"Error: Column '{column}' not found."
+            return self._last_result
+        try:
+            count = int((self._df[column].astype(str) == old_value).sum())
+            self._df[column] = self._df[column].astype(str).str.replace(old_value, new_value, regex=False)
+            self._last_result = f"Replaced {count} occurrences of '{old_value}' with '{new_value}' in '{column}'"
+        except Exception as e:
+            self._last_result = f"Error replacing values: {e}"
+        return self._last_result
+
+    def _exec_get_info(self) -> str:
+        if self._df is None:
+            return "Error: No dataset loaded."
+        info = {
+            "row_count": len(self._df),
+            "duplicate_count": int(self._df.duplicated().sum()),
+            "columns": [],
+        }
+        for col in self._df.columns:
+            col_info = {
+                "name": col,
+                "dtype": str(self._df[col].dtype),
+                "null_count": int(self._df[col].isnull().sum()),
+                "unique_count": int(self._df[col].nunique()),
+                "sample_values": [str(v) for v in self._df[col].dropna().head(3).tolist()],
+            }
+            info["columns"].append(col_info)
+        return json.dumps(info, indent=2)
+
+    # ------------------------------------------------------------------
+    # Environment API
+    # ------------------------------------------------------------------
+
+    def _get_observation_dict(self) -> Dict[str, Any]:
+        """Build observation data from current state."""
+        if self._df is None:
+            return {
+                "columns": [],
+                "row_count": 0,
+                "duplicate_count": 0,
+                "task_description": "",
+                "last_action_result": self._last_result,
+                "progress": 0.0,
+            }
+        columns_info = []
+        for col in self._df.columns:
+            columns_info.append({
+                "name": col,
+                "dtype": str(self._df[col].dtype),
+                "null_count": int(self._df[col].isnull().sum()),
+                "unique_count": int(self._df[col].nunique()),
+                "sample_values": [str(v) for v in self._df[col].dropna().head(3).tolist()],
+            })
+
+        progress = 0.0
+        if self._task and self._target is not None:
+            progress = self._task.grade(self._df, self._target)
+
+        return {
+            "columns": columns_info,
+            "row_count": len(self._df),
+            "duplicate_count": int(self._df.duplicated().sum()),
+            "task_description": self._task.description if self._task else "",
+            "last_action_result": self._last_result,
+            "progress": round(min(max(progress, 0.0), 1.0), 4),
+        }
+
+    def reset(
+        self,
+        seed: Optional[int] = None,
+        episode_id: Optional[str] = None,
+        **kwargs: Any,
+    ) -> Observation:
+        """Reset environment with a messy dataset for the configured task."""
+        task_name = kwargs.get("task", os.getenv("CSV_CLEANER_TASK", "fix_column_types"))
+        actual_seed = seed if seed is not None else 42
+
+        if task_name not in TASKS:
+            available = list(TASKS.keys())
+            return Observation(
+                done=True,
+                reward=0.0,
+                metadata={"error": f"Unknown task '{task_name}'. Available: {available}"},
+            )
+
+        self._task = TASKS[task_name]
+        self._df = self._task.generate_messy(actual_seed)
+        self._target = self._task.generate_target(actual_seed)
+        self._done = False
+        self._last_result = "Environment ready. Use get_dataset_info to see the current state."
+        self._prev_score = self._task.grade(self._df, self._target)
+        self._state = State(
+            episode_id=episode_id or str(uuid4()),
+            step_count=0,
+        )
+
+        obs_data = self._get_observation_dict()
+        return Observation(
+            done=False,
+            reward=0.0,
+            metadata={
+                "status": "ready",
+                "task": task_name,
+                "difficulty": self._task.difficulty,
+                "max_steps": self._task.max_steps,
+                "checklist": self._task.checklist,
+                **obs_data,
+            },
+        )
+
+    def _step_impl(
+        self,
+        action: Action,
+        timeout_s: Optional[float] = None,
+        **kwargs: Any,
+    ) -> Observation:
+        """Handle non-MCP actions (returns error — use MCP tools instead)."""
+        return Observation(
+            done=False,
+            reward=0.0,
+            metadata={
+                "error": f"Unknown action type: {type(action).__name__}. "
+                "Use MCP tools (get_dataset_info, cast_column, fill_missing, etc.)",
+            },
+        )
+
+    def step(
+        self,
+        action: Action,
+        timeout_s: Optional[float] = None,
+        **kwargs: Any,
+    ) -> Observation:
+        """Execute a step. Increments step count, computes reward."""
+        self._state.step_count += 1
+
+        # Let MCPEnvironment handle tool dispatch
+        obs = super().step(action, timeout_s=timeout_s, **kwargs)
+
+        # Compute reward based on progress delta
+        reward = 0.0
+        done = False
+        if self._task and self._target is not None and self._df is not None:
+            current_score = self._task.grade(self._df, self._target)
+            reward = max(0.0, current_score - self._prev_score)
+            self._prev_score = current_score
+
+            # Check if done (target reached or max steps exceeded)
+            if current_score >= 0.95:
+                done = True
+                reward += 0.1  # bonus for completing
+            elif self._state.step_count >= self._task.max_steps:
+                done = True
+
+            self._done = done
+
+        # Inject our reward/done into the observation
+        obs.reward = round(reward, 4)
+        obs.done = done
+        if obs.metadata is None:
+            obs.metadata = {}
+        obs.metadata.update(self._get_observation_dict())
+
+        return obs
+
+    async def step_async(
+        self,
+        action: Action,
+        timeout_s: Optional[float] = None,
+        **kwargs: Any,
+    ) -> Observation:
+        """Async step used by the WebSocket handler."""
+        self._state.step_count += 1
+        obs = await super().step_async(action, timeout_s=timeout_s, **kwargs)
+
+        reward = 0.0
+        done = False
+        if self._task and self._target is not None and self._df is not None:
+            current_score = self._task.grade(self._df, self._target)
+            reward = max(0.0, current_score - self._prev_score)
+            self._prev_score = current_score
+            if current_score >= 0.95:
+                done = True
+                reward += 0.1
+            elif self._state.step_count >= self._task.max_steps:
+                done = True
+            self._done = done
+
+        obs.reward = round(reward, 4)
+        obs.done = done
+        if obs.metadata is None:
+            obs.metadata = {}
+        obs.metadata.update(self._get_observation_dict())
+        return obs
+
+    @property
+    def state(self) -> State:
+        """Get current environment state."""
+        return self._state

server/tasks.py

@@ -0,0 +1,350 @@
+"""
+Task definitions for the CSV Cleaner Environment.
+
+Each task generates a deterministic messy dataset (given a seed) and defines
+a target clean dataset plus a grading function that returns a score in [0, 1].
+"""
+
+import random
+from dataclasses import dataclass, field
+from typing import Callable, Dict, List, Optional
+
+import pandas as pd
+
+
+@dataclass
+class TaskDefinition:
+    """Definition of a single cleaning task."""
+
+    name: str
+    description: str
+    difficulty: str  # easy, medium, hard
+    max_steps: int
+    generate_messy: Callable[[int], pd.DataFrame]
+    generate_target: Callable[[int], pd.DataFrame]
+    grade: Callable[[pd.DataFrame, pd.DataFrame], float]
+    checklist: List[str] = field(default_factory=list)
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+def _score_column_types(current: pd.DataFrame, target: pd.DataFrame) -> float:
+    """Score how many column types match the target."""
+    if current.empty or target.empty:
+        return 0.0
+    matching = 0
+    total = 0
+    for col in target.columns:
+        if col in current.columns:
+            total += 1
+            # Compare dtype kind (i=int, f=float, O=object, M=datetime)
+            if current[col].dtype.kind == target[col].dtype.kind:
+                matching += 1
+        else:
+            total += 1
+    return matching / total if total > 0 else 0.0
+
+
+def _score_null_counts(current: pd.DataFrame, target: pd.DataFrame) -> float:
+    """Score how close null counts are to target."""
+    if current.empty or target.empty:
+        return 0.0
+    scores = []
+    for col in target.columns:
+        if col in current.columns:
+            target_nulls = target[col].isnull().sum()
+            current_nulls = current[col].isnull().sum()
+            if target_nulls == 0:
+                scores.append(1.0 if current_nulls == 0 else max(0.0, 1.0 - current_nulls / max(len(current), 1)))
+            else:
+                scores.append(1.0 - min(1.0, abs(current_nulls - target_nulls) / max(len(current), 1)))
+    return sum(scores) / len(scores) if scores else 0.0
+
+
+def _score_duplicates(current: pd.DataFrame, target: pd.DataFrame) -> float:
+    """Score duplicate removal progress."""
+    target_dups = target.duplicated().sum()
+    current_dups = current.duplicated().sum()
+    if target_dups == 0:
+        if current_dups == 0:
+            return 1.0
+        return max(0.0, 1.0 - current_dups / max(len(current), 1))
+    return 1.0 - min(1.0, abs(current_dups - target_dups) / max(len(current), 1))
+
+
+def _score_row_count(current: pd.DataFrame, target: pd.DataFrame) -> float:
+    """Score how close row count is to target."""
+    if len(target) == 0:
+        return 1.0 if len(current) == 0 else 0.0
+    diff = abs(len(current) - len(target))
+    return max(0.0, 1.0 - diff / max(len(target), 1))
+
+
+def _score_column_names(current: pd.DataFrame, target: pd.DataFrame) -> float:
+    """Score how many column names match the target."""
+    target_cols = set(target.columns)
+    current_cols = set(current.columns)
+    if not target_cols:
+        return 1.0
+    return len(target_cols & current_cols) / len(target_cols)
+
+
+# ---------------------------------------------------------------------------
+# Task 1: Easy — Fix Column Types
+# ---------------------------------------------------------------------------
+
+def _easy_generate_messy(seed: int) -> pd.DataFrame:
+    """Generate a dataset with wrong column types."""
+    rng = random.Random(seed)
+    n = 20
+    data = {
+        "employee_id": [str(rng.randint(1000, 9999)) for _ in range(n)],
+        "name": [rng.choice(["Alice", "Bob", "Charlie", "Diana", "Eve", "Frank", "Grace", "Hank"]) for _ in range(n)],
+        "age": [str(rng.randint(22, 65)) for _ in range(n)],
+        "salary": [f"{rng.uniform(30000, 120000):.2f}" for _ in range(n)],
+        "join_date": [f"2{rng.randint(0, 0)}2{rng.randint(0, 4)}-{rng.randint(1, 12):02d}-{rng.randint(1, 28):02d}" for _ in range(n)],
+        "department": [rng.choice(["Engineering", "Sales", "Marketing", "HR", "Finance"]) for _ in range(n)],
+    }
+    return pd.DataFrame(data)
+
+
+def _easy_generate_target(seed: int) -> pd.DataFrame:
+    """Generate the target clean dataset for task 1."""
+    df = _easy_generate_messy(seed)
+    df["employee_id"] = df["employee_id"].astype(int)
+    df["age"] = df["age"].astype(int)
+    df["salary"] = df["salary"].astype(float)
+    df["join_date"] = pd.to_datetime(df["join_date"])
+    return df
+
+
+def _easy_grade(current: pd.DataFrame, target: pd.DataFrame) -> float:
+    """Grade task 1: type matching is the primary objective."""
+    type_score = _score_column_types(current, target)
+    row_score = _score_row_count(current, target)
+    return 0.8 * type_score + 0.2 * row_score
+
+
+# ---------------------------------------------------------------------------
+# Task 2: Medium — Clean Missing Values + Remove Duplicates
+# ---------------------------------------------------------------------------
+
+def _medium_generate_messy(seed: int) -> pd.DataFrame:
+    """Generate a dataset with missing values and duplicates."""
+    rng = random.Random(seed)
+    n = 30
+    base_data = []
+    for i in range(n):
+        row = {
+            "product_id": i + 1,
+            "product_name": rng.choice(["Widget A", "Widget B", "Gadget X", "Gadget Y", "Tool M", "Tool N"]),
+            "category": rng.choice(["Electronics", "Hardware", "Software", "Accessories"]),
+            "price": round(rng.uniform(5.0, 500.0), 2),
+            "stock": rng.randint(0, 1000),
+        }
+        # Inject nulls
+        if rng.random() < 0.2:
+            row["price"] = None
+        if rng.random() < 0.15:
+            row["category"] = None
+        if rng.random() < 0.1:
+            row["stock"] = None
+        base_data.append(row)
+
+    # Inject duplicates (copy ~5 random rows)
+    for _ in range(5):
+        idx = rng.randint(0, len(base_data) - 1)
+        base_data.append(base_data[idx].copy())
+
+    rng.shuffle(base_data)
+    return pd.DataFrame(base_data)
+
+
+def _medium_generate_target(seed: int) -> pd.DataFrame:
+    """Generate the target clean dataset for task 2."""
+    df = _medium_generate_messy(seed)
+    # Fill missing price with median
+    median_price = df["price"].median()
+    df["price"] = df["price"].fillna(median_price)
+    # Fill missing category with mode
+    mode_cat = df["category"].mode()[0] if not df["category"].mode().empty else "Unknown"
+    df["category"] = df["category"].fillna(mode_cat)
+    # Fill missing stock with 0
+    df["stock"] = df["stock"].fillna(0).astype(int)
+    # Drop duplicates
+    df = df.drop_duplicates().reset_index(drop=True)
+    return df
+
+
+def _medium_grade(current: pd.DataFrame, target: pd.DataFrame) -> float:
+    """Grade task 2: null handling + duplicate removal."""
+    null_score = _score_null_counts(current, target)
+    dup_score = _score_duplicates(current, target)
+    row_score = _score_row_count(current, target)
+    return 0.4 * null_score + 0.35 * dup_score + 0.25 * row_score
+
+
+# ---------------------------------------------------------------------------
+# Task 3: Hard — Full Pipeline
+# ---------------------------------------------------------------------------
+
+def _hard_generate_messy(seed: int) -> pd.DataFrame:
+    """Generate a dataset needing the full cleaning pipeline."""
+    rng = random.Random(seed)
+    n = 40
+    base_data = []
+    for i in range(n):
+        row = {
+            "cust_id": str(rng.randint(10000, 99999)),
+            "  Full Name ": rng.choice([
+                "  John Smith ", "Alice Johnson", " Bob Williams  ",
+                "Charlie Brown", "  Diana Ross", "Eve Davis  ",
+                "Frank Miller", " Grace Lee ",
+            ]),
+            "email_addr": rng.choice([
+                "john@example.com", "alice@test.com", "bob@demo.com",
+                "charlie@sample.org", "diana@mail.com", "INVALID",
+                "eve@test.com", "frank@example.com",
+            ]),
+            "purchase_amt": f"${rng.uniform(10, 5000):.2f}" if rng.random() > 0.15 else str(round(rng.uniform(10, 5000), 2)),
+            "rating": str(rng.randint(1, 5)) if rng.random() > 0.1 else None,
+            "signup_date": f"2{rng.randint(0, 0)}2{rng.randint(0, 4)}-{rng.randint(1, 12):02d}-{rng.randint(1, 28):02d}" if rng.random() > 0.1 else None,
+            "status": rng.choice(["active", "Active", "ACTIVE", "inactive", "Inactive", "INACTIVE"]),
+        }
+        # Inject some nulls
+        if rng.random() < 0.12:
+            row["email_addr"] = None
+        base_data.append(row)
+
+    # Inject duplicates
+    for _ in range(6):
+        idx = rng.randint(0, len(base_data) - 1)
+        base_data.append(base_data[idx].copy())
+
+    rng.shuffle(base_data)
+    return pd.DataFrame(base_data)
+
+
+def _hard_generate_target(seed: int) -> pd.DataFrame:
+    """Generate the target clean dataset for task 3."""
+    df = _hard_generate_messy(seed)
+    # Rename columns
+    df = df.rename(columns={
+        "  Full Name ": "full_name",
+        "email_addr": "email",
+        "purchase_amt": "purchase_amount",
+        "signup_date": "signup_date",
+        "cust_id": "customer_id",
+    })
+    # Strip whitespace from full_name
+    df["full_name"] = df["full_name"].str.strip()
+    # Cast customer_id to int
+    df["customer_id"] = df["customer_id"].astype(int)
+    # Clean purchase_amount: remove $ and cast to float
+    df["purchase_amount"] = df["purchase_amount"].astype(str).str.replace("$", "", regex=False).astype(float)
+    # Cast rating to int/float, fill missing with median
+    df["rating"] = pd.to_numeric(df["rating"], errors="coerce")
+    median_rating = df["rating"].median()
+    df["rating"] = df["rating"].fillna(median_rating).astype(int)
+    # Normalize status to lowercase
+    df["status"] = df["status"].str.lower()
+    # Fill missing signup_date with a sentinel
+    df["signup_date"] = pd.to_datetime(df["signup_date"], errors="coerce")
+    # Fill missing email
+    df["email"] = df["email"].fillna("unknown@example.com")
+    # Filter out INVALID emails
+    df = df[df["email"] != "INVALID"].reset_index(drop=True)
+    # Drop duplicates
+    df = df.drop_duplicates().reset_index(drop=True)
+    return df
+
+
+def _hard_grade(current: pd.DataFrame, target: pd.DataFrame) -> float:
+    """Grade task 3: full pipeline."""
+    name_score = _score_column_names(current, target)
+    type_score = _score_column_types(current, target)
+    null_score = _score_null_counts(current, target)
+    dup_score = _score_duplicates(current, target)
+    row_score = _score_row_count(current, target)
+    return (0.15 * name_score + 0.25 * type_score + 0.25 * null_score +
+            0.15 * dup_score + 0.20 * row_score)
+
+
+# ---------------------------------------------------------------------------
+# Task Registry
+# ---------------------------------------------------------------------------
+
+TASKS: Dict[str, TaskDefinition] = {
+    "fix_column_types": TaskDefinition(
+        name="fix_column_types",
+        description=(
+            "Fix column types in an employee dataset. Columns employee_id, age, "
+            "salary, and join_date are stored as strings but should be int, int, "
+            "float, and datetime respectively. Cast them to the correct types."
+        ),
+        difficulty="easy",
+        max_steps=10,
+        generate_messy=_easy_generate_messy,
+        generate_target=_easy_generate_target,
+        grade=_easy_grade,
+        checklist=[
+            "Cast employee_id from string to int",
+            "Cast age from string to int",
+            "Cast salary from string to float",
+            "Cast join_date from string to datetime",
+        ],
+    ),
+    "clean_missing_duplicates": TaskDefinition(
+        name="clean_missing_duplicates",
+        description=(
+            "Clean a product inventory dataset. Fill missing price values with the "
+            "median, fill missing category with the mode, fill missing stock with 0, "
+            "then remove all duplicate rows."
+        ),
+        difficulty="medium",
+        max_steps=15,
+        generate_messy=_medium_generate_messy,
+        generate_target=_medium_generate_target,
+        grade=_medium_grade,
+        checklist=[
+            "Fill missing price with median",
+            "Fill missing category with mode",
+            "Fill missing stock with 0",
+            "Remove duplicate rows",
+        ],
+    ),
+    "full_pipeline": TaskDefinition(
+        name="full_pipeline",
+        description=(
+            "Perform a full cleaning pipeline on a customer dataset: "
+            "(1) Rename '  Full Name ' to 'full_name' and 'email_addr' to 'email' "
+            "and 'purchase_amt' to 'purchase_amount' and 'cust_id' to 'customer_id'. "
+            "(2) Strip whitespace from full_name. "
+            "(3) Cast customer_id to int. "
+            "(4) Remove '$' from purchase_amount and cast to float. "
+            "(5) Cast rating to int, fill missing with median. "
+            "(6) Normalize status to lowercase. "
+            "(7) Fill missing email with 'unknown@example.com'. "
+            "(8) Filter out rows where email is 'INVALID'. "
+            "(9) Remove duplicate rows."
+        ),
+        difficulty="hard",
+        max_steps=20,
+        generate_messy=_hard_generate_messy,
+        generate_target=_hard_generate_target,
+        grade=_hard_grade,
+        checklist=[
+            "Rename columns to clean names",
+            "Strip whitespace from full_name",
+            "Cast customer_id to int",
+            "Clean and cast purchase_amount to float",
+            "Cast rating to int, fill missing with median",
+            "Normalize status to lowercase",
+            "Fill missing email",
+            "Filter out INVALID emails",
+            "Remove duplicate rows",
+        ],
+    ),
+}