Deploy dataops-env Space

.dataops_policy_cache.json

@@ -0,0 +1,247 @@
+{
+  "patterns": {
+    "0fa3f35b7f17ba183b55874ed34f34b9dfd42a96": {
+      "actions": {
+        "fill_missing(row_id=34, column='city', value='Denver')": {
+          "attempts": 1,
+          "cumulative_reward": 0.47,
+          "failures": 0,
+          "last_error": null,
+          "progresses": 0,
+          "successes": 1
+        }
+      }
+    },
+    "11ef7431147857285b2046abba4c938d39542f56": {
+      "actions": {
+        "normalize_column(column='email')": {
+          "attempts": 1,
+          "cumulative_reward": 0.2738,
+          "failures": 0,
+          "last_error": null,
+          "progresses": 1,
+          "successes": 0
+        }
+      }
+    },
+    "468a9ffd1e161a3b120e50a2d283b5a2bea91b78": {
+      "actions": {
+        "delete_row(row_id=23)": {
+          "attempts": 1,
+          "cumulative_reward": 0.13,
+          "failures": 0,
+          "last_error": null,
+          "progresses": 1,
+          "successes": 0
+        }
+      }
+    },
+    "47741633a233a9bfecedd98ec5d8a0cafed7aa58": {
+      "actions": {
+        "normalize_column(column='city')": {
+          "attempts": 1,
+          "cumulative_reward": 0.4363,
+          "failures": 0,
+          "last_error": null,
+          "progresses": 0,
+          "successes": 1
+        }
+      }
+    },
+    "84f1a657f98b24e5c81396ebede79cb10b4bd759": {
+      "actions": {
+        "remove_duplicate(row_id=33)": {
+          "attempts": 1,
+          "cumulative_reward": 0.43,
+          "failures": 0,
+          "last_error": null,
+          "progresses": 1,
+          "successes": 0
+        }
+      }
+    },
+    "a4f3a4b48581add24c1541d266ca386ca1048501": {
+      "actions": {
+        "normalize_column(column='name')": {
+          "attempts": 1,
+          "cumulative_reward": 0.2363,
+          "failures": 0,
+          "last_error": null,
+          "progresses": 1,
+          "successes": 0
+        }
+      }
+    },
+    "b2557fa133ca411bea9c1d016bb1022ef521dd12": {
+      "actions": {
+        "delete_row(row_id=26)": {
+          "attempts": 1,
+          "cumulative_reward": 0.2657,
+          "failures": 0,
+          "last_error": null,
+          "progresses": 0,
+          "successes": 1
+        }
+      }
+    },
+    "bac5c03eaec86e5eca3602c18787e7b8dd71ce77": {
+      "actions": {
+        "fill_missing(row_id=35, column='email', value='peak.systems@example.com')": {
+          "attempts": 1,
+          "cumulative_reward": 0.24,
+          "failures": 0,
+          "last_error": null,
+          "progresses": 1,
+          "successes": 0
+        }
+      }
+    },
+    "c9a1695ddb1fd7a56fcd89d0855f3e6c3cc0539f": {
+      "actions": {
+        "remove_duplicate(row_id=13)": {
+          "attempts": 1,
+          "cumulative_reward": 0.3737,
+          "failures": 0,
+          "last_error": null,
+          "progresses": 1,
+          "successes": 0
+        }
+      }
+    },
+    "e133231c2a2fe412ce1265f027ce61ff2616d497": {
+      "actions": {
+        "remove_duplicate(row_id=22)": {
+          "attempts": 1,
+          "cumulative_reward": 0.3443,
+          "failures": 0,
+          "last_error": null,
+          "progresses": 1,
+          "successes": 0
+        }
+      }
+    }
+  },
+  "states": {
+    "0cc9851ce3eac00c3a8ba813a0bbe5086b6b28c7": {
+      "actions": {
+        "fill_missing(row_id=35, column='email', value='peak.systems@example.com')": {
+          "attempts": 1,
+          "cumulative_reward": 0.24,
+          "failures": 0,
+          "last_error": null,
+          "progresses": 1,
+          "successes": 0
+        }
+      }
+    },
+    "338f581f2bb3d33b1ca9113338487576d7221075": {
+      "actions": {
+        "remove_duplicate(row_id=13)": {
+          "attempts": 1,
+          "cumulative_reward": 0.3737,
+          "failures": 0,
+          "last_error": null,
+          "progresses": 1,
+          "successes": 0
+        }
+      }
+    },
+    "3641040a720ad30712929dfc63af936f954c7048": {
+      "actions": {
+        "delete_row(row_id=26)": {
+          "attempts": 1,
+          "cumulative_reward": 0.2657,
+          "failures": 0,
+          "last_error": null,
+          "progresses": 0,
+          "successes": 1
+        }
+      }
+    },
+    "4cdcb94d06f82ed5c380e36d4737c98095c1d8e5": {
+      "actions": {
+        "fill_missing(row_id=34, column='city', value='Denver')": {
+          "attempts": 1,
+          "cumulative_reward": 0.47,
+          "failures": 0,
+          "last_error": null,
+          "progresses": 0,
+          "successes": 1
+        }
+      }
+    },
+    "72db3bd5a5ba3a4f64815e20b59ccf3b9ce6b2ad": {
+      "actions": {
+        "remove_duplicate(row_id=33)": {
+          "attempts": 1,
+          "cumulative_reward": 0.43,
+          "failures": 0,
+          "last_error": null,
+          "progresses": 1,
+          "successes": 0
+        }
+      }
+    },
+    "a3e1f7d06da6a2a3bfc3ca13a3a9ae20ff116705": {
+      "actions": {
+        "delete_row(row_id=23)": {
+          "attempts": 1,
+          "cumulative_reward": 0.13,
+          "failures": 0,
+          "last_error": null,
+          "progresses": 1,
+          "successes": 0
+        }
+      }
+    },
+    "be2f2def033a20dab83499d31cbe7ac8ba296d8e": {
+      "actions": {
+        "normalize_column(column='name')": {
+          "attempts": 1,
+          "cumulative_reward": 0.2363,
+          "failures": 0,
+          "last_error": null,
+          "progresses": 1,
+          "successes": 0
+        }
+      }
+    },
+    "c02de8edbfcfa72f061a9ac1827b98c070f1208e": {
+      "actions": {
+        "remove_duplicate(row_id=22)": {
+          "attempts": 1,
+          "cumulative_reward": 0.3443,
+          "failures": 0,
+          "last_error": null,
+          "progresses": 1,
+          "successes": 0
+        }
+      }
+    },
+    "d30ee334a5d05193f192a491cf4e069c1b494b21": {
+      "actions": {
+        "normalize_column(column='city')": {
+          "attempts": 1,
+          "cumulative_reward": 0.4363,
+          "failures": 0,
+          "last_error": null,
+          "progresses": 0,
+          "successes": 1
+        }
+      }
+    },
+    "f1c2df4babd9a79a88196afded5618bb0ce3d72b": {
+      "actions": {
+        "normalize_column(column='email')": {
+          "attempts": 1,
+          "cumulative_reward": 0.2738,
+          "failures": 0,
+          "last_error": null,
+          "progresses": 1,
+          "successes": 0
+        }
+      }
+    }
+  },
+  "version": 1
+}

.dockerignore

@@ -0,0 +1,13 @@
+__pycache__/
+*.pyc
+*.pyo
+*.pyd
+.git/
+.venv/
+venv/
+dist/
+build/
+.pytest_cache/
+.mypy_cache/
+agent-tools/
+terminals/

Dockerfile

@@ -0,0 +1,27 @@
+FROM python:3.10-slim
+
+ENV PYTHONDONTWRITEBYTECODE=1
+ENV PYTHONUNBUFFERED=1
+ENV PYTHONPATH=/app
+ENV PORT=7860
+
+WORKDIR /app
+
+RUN apt-get update && apt-get install -y \
+    build-essential \
+    curl \
+    && rm -rf /var/lib/apt/lists/*
+
+COPY requirements.txt .
+RUN pip install --no-cache-dir --upgrade pip && \
+    pip install --no-cache-dir -r requirements.txt
+
+COPY . .
+
+#Non-root user
+RUN useradd -m appuser
+USER appuser
+
+EXPOSE 7860
+
+CMD ["sh", "-c", "uvicorn server.app:app --host 0.0.0.0 --port ${PORT:-7860}"]

README.md

@@ -1,10 +1,344 @@
 ---
+
 title: Dataops Env
 emoji: 📊
 colorFrom: indigo
 colorTo: gray
 sdk: docker
+app_port: 7860
 pinned: false
+
 ---
 
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+# `dataops-env`
+
+`dataops-env` is an OpenEnv benchmark for training and evaluating agents on
+multi-step data operations work. Instead of a single obvious cleanup action, an
+agent must inspect messy business tables, choose corrective actions in the right
+order, preserve valid-but-unusual records, and know when the table is truly
+ready for validation.
+
+It exposes the standard `reset()`, `step(action)`, and `state()` interface,
+ships with a production-ready FastAPI server and Docker image, and includes a
+reproducible OpenAI-compatible baseline runner.
+
+## Benchmark Purpose
+
+Many toy data-cleaning tasks reward shallow pattern matching. Real operational
+data work is harder:
+
+- duplicates may be safe to remove, but conflicting rows require judgment
+- some malformed values should be normalized, while unusual valid values must be preserved
+- deletion is often the riskiest action, not the default fix
+- agents need partial credit for progress, but strong penalties for repeated mistakes
+
+`dataops-env` is designed to capture those decisions in a compact benchmark that
+is still easy to run, validate, and deploy in the OpenEnv ecosystem.
+
+## Why It Feels Real
+
+The environment models common enterprise data quality problems:
+
+- exact duplicates in customer or vendor master data
+- missing required fields
+- inconsistent casing in names and locations
+- invalid email and phone formats
+- conflicting records for the same real-world entity
+- uniqueness constraints such as shared-email violations
+- trap rows that look suspicious but are actually valid
+
+Agents are rewarded for minimal corrective behavior and punished for destructive
+or repetitive actions. That makes the environment useful for both learning and
+evaluation.
+
+## Task Families
+
+The benchmark keeps the hackathon-friendly `easy`, `medium`, and `hard` task
+structure, while each family now contains deterministic variants so policies
+cannot overfit a single table.
+
+1. `easy`
+   Remove duplicates and fill missing required fields.
+2. `medium`
+   Remove duplicates, normalize casing, and repair invalid emails.
+3. `hard`
+   Resolve conflicts, enforce unique-email constraints, fix invalid formats,
+   and preserve valid trap rows.
+
+Each task definition includes:
+
+- `goal`
+- `difficulty`
+- `variant_id`
+- `required_columns`
+- `hidden_issues`
+- `constraints`
+- `expected_outcome`
+- `max_steps`
+
+## Learning Signals
+
+The environment provides both dense rewards and a deterministic final score:
+
+- partial rewards for duplicate removal, normalization, and filling missing values
+- step costs and no-progress penalties to discourage random actions
+- escalating penalties for repeated mistakes
+- destructive-action penalties for harmful deletions
+- proactive hints after recurring failures
+- final task scoring on a strict `0.0` to `1.0` scale
+
+The final task score and the visible validation failures are produced from the
+same explicit rule set, reducing mismatch between what the agent sees and how it
+is ultimately judged.
+
+## Action Space
+
+Agents interact with the environment through a typed `Action` object.
+
+Supported action types:
+
+- `remove_duplicate`
+  Remove one row from an exact duplicate group. Can be called with an explicit
+  `row_id`, or the environment can choose the default duplicate target.
+- `fill_missing`
+  Fill a missing field on a target row. Requires `column` and `value`, and may
+  also include `row_id`.
+- `normalize_column`
+  Apply deterministic normalization to a supported column such as `name`,
+  `city`, `email`, or `phone`.
+- `delete_row`
+  Delete a row when doing so resolves a structural issue like a conflict or a
+  uniqueness violation. Requires `row_id`.
+- `validate`
+  Signal that the agent believes the table is ready for completion.
+- `noop`
+  Explicitly take no action. This is allowed but penalized when unresolved
+  issues remain.
+
+Typed action schema:
+
+- `action_id: Optional[str]`
+- `action_type: Literal["remove_duplicate", "fill_missing", "normalize_column", "delete_row", "validate", "noop"]`
+- `column: Optional[str]`
+- `row_id: Optional[int]`
+- `value: Optional[str]`
+
+Validation rules:
+
+- `delete_row` requires `row_id`
+- `normalize_column` requires `column`
+- `fill_missing` requires `column` and `value`
+
+Example actions:
+
+```json
+{"action_id":"step-001","action_type":"remove_duplicate","row_id":33}
+{"action_id":"step-002","action_type":"fill_missing","row_id":35,"column":"email","value":"peak.systems@example.com"}
+{"action_id":"step-003","action_type":"normalize_column","column":"email"}
+{"action_id":"step-004","action_type":"validate"}
+```
+
+## Observation Space
+
+The environment returns a typed `Observation` object after `reset()` and each
+call to `step()`.
+
+Observation fields:
+
+- `goal: str`
+  Natural-language description of what the agent should accomplish.
+- `table: List[Dict[str, Any]]`
+  Current JSON-serializable table snapshot.
+- `issues: List[str]`
+  Human-readable unresolved issues and validation failures.
+- `history: List[str]`
+  Ordered record of previous actions/events in the current episode.
+- `mistakes: Dict[str, int]`
+  Counts of repeated mistake categories tracked during the episode.
+- `hints: List[str]`
+  Proactive or reactive guidance derived from issue state and prior failures.
+- `progress: float`
+  Normalized progress estimate in `[0.0, 1.0]`.
+- `steps_remaining: int`
+  Number of remaining actions before the episode terminates.
+
+Example observation shape:
+
+```json
+{
+  "goal": "Normalize the dataset by fixing casing, removing duplicates, and correcting invalid email formats.",
+  "table": [
+    {"row_id": 10, "customer_id": "C100", "name": "jane miller", "city": "new york", "email": "jane.miller@example.com"}
+  ],
+  "issues": [
+    "Rows 11 and 13 are duplicates and only one should remain."
+  ],
+  "history": [],
+  "mistakes": {},
+  "hints": [],
+  "progress": 0.0,
+  "steps_remaining": 9
+}
+```
+
+## Expected Agent Behavior
+
+A strong agent should behave roughly like this:
+
+1. inspect the visible table and unresolved issues
+2. remove safe duplicates first
+3. repair missing or malformed values without over-editing valid rows
+4. resolve structural conflicts carefully, especially in hard tasks
+5. validate only when the remaining issue list is empty
+
+Example successful baseline trace:
+
+```text
+[START] task=medium env=dataops-env model=your-model
+[STEP] step=1 action=remove_duplicate(row_id=13) reward=0.37 done=false error=null
+[STEP] step=2 action=normalize_column(column='email') reward=0.27 done=false error=null
+[STEP] step=3 action=normalize_column(column='name') reward=0.24 done=false error=null
+[STEP] step=4 action=normalize_column(column='city') reward=0.44 done=true error=null
+[END] success=true steps=4 rewards=0.37,0.27,0.24,0.44
+```
+
+## Project Layout
+
+- `env.py`: core `DataOpsEnv` implementation
+- `task.py`: task families and deterministic variants
+- `models.py`: typed `Action`, `Observation`, and `Reward` contracts
+- `grader.py`: dense rewards, explicit validation checks, and final task scoring
+- `server/app.py`: FastAPI runtime API
+- `inference.py`: hybrid heuristic/model baseline runner
+- `openenv.yaml`: OpenEnv metadata and task registration
+- `pyproject.toml`: package metadata and server script entry point
+- `Dockerfile`: production container image
+
+## Local Setup
+
+```bash
+pip install -r requirements.txt
+openenv validate
+```
+
+Run the FastAPI server:
+
+```bash
+python -m server.app
+```
+
+By default, the local server runs on port `8000`.
+
+Or use the packaged entry point:
+
+```bash
+server
+```
+
+## API
+
+Health check:
+
+```bash
+curl http://localhost:8000/health
+```
+
+Create a session with an optional seed and task selection:
+
+```bash
+curl -X POST http://localhost:8000/reset \
+  -H "Content-Type: application/json" \
+  -d '{"seed": 0, "task_name": "easy"}'
+```
+
+Step the environment:
+
+```bash
+curl -X POST "http://localhost:8000/step" \
+  -H "Content-Type: application/json" \
+  -d '{"action_id":"step-001","action_type":"validate"}'
+```
+
+Read internal state:
+
+```bash
+curl "http://localhost:8000/state"
+```
+
+## Baseline Inference
+
+The baseline runner now combines deterministic local planning with optional
+model arbitration. The local planner proposes ranked candidate actions from the
+visible table state, and the model is constrained to choose only from those
+candidates. This avoids many common failure modes such as invalid actions,
+repeated no-op loops, and reckless deletion choices.
+
+Run it with an OpenAI-compatible endpoint:
+
+```bash
+set HF_TOKEN=your_token
+set MODEL_NAME=your_model
+set API_BASE_URL=https://router.huggingface.co/v1
+python inference.py
+```
+
+Key properties:
+
+- strict `[START]`, `[STEP]`, and `[END]` output formatting
+- fixed task ordering for reproducibility
+- retry logic for invalid or blocked model suggestions
+- strong heuristic fallback when the model is unavailable
+- action filtering based on prior no-progress or errorful behavior
+
+## Docker
+
+Build:
+
+```bash
+docker build -t dataops-env .
+```
+
+Run locally:
+
+```bash
+docker run -p 8000:8000 dataops-env
+```
+
+## Hugging Face Spaces Notes
+
+For Hugging Face `Docker` Spaces, the container should normally listen on port
+`7860`, or the Space must be explicitly configured to expect a different
+internal port.
+
+If you keep the current container on port `8000`, make sure your Space is
+configured with:
+
+```yaml
+app_port: 8000
+```
+
+If you want the simplest Hugging Face Spaces setup, change the container to use
+port `7860` instead:
+
+```dockerfile
+EXPOSE 7860
+CMD ["uvicorn", "server.app:app", "--host", "0.0.0.0", "--port", "7860"]
+```
+
+Then local Docker testing would become:
+
+```bash
+docker run -p 7860:7860 dataops-env
+curl http://localhost:7860/health
+```
+
+## Submission Notes
+
+- `openenv validate` passes
+- the server and Docker image run successfully
+- the packaged benchmark supports multi-mode deployment
+- the default baseline now completes the public task families deterministically
+
+Leaderboard performance will still depend on the quality of the external model,
+but the repository is now structured and documented like a serious benchmark
+submission rather than a starter scaffold.

env.py

@@ -0,0 +1,756 @@
+"""OpenEnv environment entrypoint for ``dataops-gym``.
+
+This module is responsible for declaring top-level environment metadata,
+configuration wiring, and lifecycle integration points for the OpenEnv runtime.
+"""
+
+from __future__ import annotations
+
+from copy import deepcopy
+import random
+import re
+from typing import Any, Dict, Iterable, List, Mapping, MutableMapping, Optional, Tuple
+
+from grader import grade_step_details, grade_task_result, task_failure_messages
+from models import Action, Observation
+from task import (
+    HiddenIssue,
+    TaskDefinition,
+    easy_cleaning_task,
+    hard_conflict_resolution_task,
+    medium_normalization_task,
+)
+
+
+EMAIL_PATTERN = re.compile(r"^[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}$")
+
+
+class DataOpsEnv:
+    """Deterministic multi-step data-cleaning environment for OpenEnv."""
+
+    def __init__(self, seed: int = 0, task_name: Optional[str] = None) -> None:
+        """Initialize the environment with deterministic task sampling."""
+
+        self._seed = seed
+        self._rng = random.Random(seed)
+        self._task_registry: List[Tuple[str, Any]] = [
+            ("easy", easy_cleaning_task),
+            ("medium", medium_normalization_task),
+            ("hard", hard_conflict_resolution_task),
+        ]
+        self._fixed_task_name = task_name
+        self._global_mistake_memory: Dict[str, int] = {}
+        self._state_data: Dict[str, Any] = {}
+
+    def reset(self) -> Observation:
+        """Load a random task, initialize episode state, and return an observation."""
+
+        task_name, task_factory = self._select_task_factory()
+        variant_count = max(1, int(getattr(task_factory, "variant_count", 1)))
+        variant_index = self._rng.randrange(variant_count)
+        task_definition = deepcopy(task_factory(variant=variant_index))
+        initial_table = deepcopy(task_definition["initial_table"])
+        initial_table_by_row_id = self._table_by_row_id(initial_table)
+
+        self._state_data = {
+            "seed": self._seed,
+            "task_name": task_name,
+            "task_variant": task_definition.get("variant_id", f"{task_name}_variant_{variant_index}"),
+            "task": task_definition,
+            "table": initial_table,
+            "history": [],
+            "mistakes": {},
+            "mistake_memory": [],
+            "hints": [],
+            "steps_taken": 0,
+            "steps_remaining": task_definition["max_steps"],
+            "done": False,
+            "last_reward_components": {},
+            "last_info": {},
+            "last_task_score": 0.0,
+            "initial_issue_count": 1,
+            "initial_table_by_row_id": initial_table_by_row_id,
+        }
+        initial_issue_count = len(self._current_issue_messages(initial_table, task_definition))
+        self._state_data["initial_issue_count"] = max(1, initial_issue_count)
+        return self._build_observation()
+
+    def step(
+        self, action: Action | Mapping[str, Any]
+    ) -> Tuple[Observation, float, bool, Dict[str, Any]]:
+        """Apply one action, score it, update state, and return a gym-style step tuple."""
+
+        if not self._state_data:
+            raise RuntimeError("Environment must be reset before calling step().")
+        if self._state_data.get("done", False):
+            raise RuntimeError("Episode is finished. Call reset() before stepping again.")
+
+        parsed_action, action_error = self._coerce_action(action)
+        task_definition: TaskDefinition = self._state_data["task"]
+        table_before = deepcopy(self._state_data["table"])
+        issues_before = self._current_issue_messages(table_before, task_definition)
+
+        result: Dict[str, Any] = {
+            "mistake_keys": [],
+            "error_type": "general",
+        }
+
+        if action_error is not None:
+            parsed_action = Action(action_type="noop")
+            result["noop"] = True
+            result["unnecessary_action"] = True
+            result["error_type"] = "invalid_action"
+            result["mistake_keys"].append("invalid_action:general")
+            history_entry = f"invalid_action({action_error})"
+        else:
+            history_entry = self._apply_action(parsed_action, result)
+
+        self._state_data["history"].append(history_entry)
+        self._state_data["steps_taken"] += 1
+        self._state_data["steps_remaining"] = max(
+            0, task_definition["max_steps"] - self._state_data["steps_taken"]
+        )
+
+        table_after = deepcopy(self._state_data["table"])
+        issues_after = self._current_issue_messages(table_after, task_definition)
+        self._populate_result_signals(
+            parsed_action,
+            table_before,
+            table_after,
+            issues_before,
+            issues_after,
+            result,
+        )
+
+        reward, components = grade_step_details(
+            self._state_data, parsed_action.model_dump(), result
+        )
+        self._record_mistake_memory(parsed_action, result)
+        self._update_hints(result, issues_after)
+
+        done = not issues_after or self._state_data["steps_remaining"] <= 0
+        self._state_data["done"] = done
+        task_score = grade_task_result(
+            task_definition, self._state_data["table"], self._state_data
+        )
+        self._state_data["last_task_score"] = task_score
+
+        observation = self._build_observation()
+        info = {
+            "task_name": self._state_data["task_name"],
+            "task_variant": self._state_data["task_variant"],
+            "difficulty": task_definition["difficulty"],
+            "reward_components": components,
+            "mistakes": deepcopy(self._state_data["mistakes"]),
+            "hints": list(self._state_data["hints"]),
+            "issues_remaining": len(issues_after),
+            "done_reason": "resolved" if not issues_after else "max_steps" if done else None,
+            "task_score": task_score,
+            "result": deepcopy(result),
+        }
+        self._state_data["last_reward_components"] = deepcopy(components)
+        self._state_data["last_info"] = deepcopy(info)
+        return observation, reward, done, info
+
+    def state(self) -> Dict[str, Any]:
+        """Return a deep copy of the internal environment state."""
+
+        return deepcopy(self._state_data)
+
+    def close(self) -> None:
+        """Release environment state for callers using explicit lifecycle cleanup."""
+
+        self._state_data = {}
+
+    def _select_task_factory(self) -> Tuple[str, Any]:
+        """Pick the configured task factory deterministically."""
+
+        if self._fixed_task_name is None:
+            return self._rng.choice(self._task_registry)
+
+        for task_name, task_factory in self._task_registry:
+            if self._fixed_task_name in {task_name, task_factory.__name__}:
+                return task_name, task_factory
+
+        raise ValueError(f"Unknown task_name: {self._fixed_task_name}")
+
+    def _coerce_action(
+        self, action: Action | Mapping[str, Any]
+    ) -> Tuple[Optional[Action], Optional[str]]:
+        """Convert user input into an ``Action`` model without raising outward."""
+
+        if isinstance(action, Action):
+            return action, None
+
+        try:
+            return Action(**dict(action)), None
+        except Exception as exc:  # pragma: no cover - defensive runtime boundary
+            return None, str(exc)
+
+    def _apply_action(self, action: Action, result: MutableMapping[str, Any]) -> str:
+        """Apply a single action to the current table and capture side effects."""
+
+        if action.action_type == "noop":
+            result["noop"] = True
+            result["mistake_keys"].append(f"{action.action_type}:noop")
+            return self._format_history(action)
+
+        if action.action_type == "remove_duplicate":
+            self._remove_duplicate(action, result)
+            return self._format_history(action)
+
+        if action.action_type == "delete_row":
+            self._delete_row(action, result)
+            return self._format_history(action)
+
+        if action.action_type == "fill_missing":
+            self._fill_missing(action, result)
+            return self._format_history(action)
+
+        if action.action_type == "normalize_column":
+            self._normalize_column(action, result)
+            return self._format_history(action)
+
+        if action.action_type == "validate":
+            return self._format_history(action)
+
+        result["unnecessary_action"] = True
+        result["error_type"] = "unsupported_action"
+        result["mistake_keys"].append(f"{action.action_type}:unsupported_action")
+        return self._format_history(action)
+
+    def _remove_duplicate(
+        self, action: Action, result: MutableMapping[str, Any]
+    ) -> None:
+        """Remove a duplicate row when the target belongs to a duplicate issue."""
+
+        duplicate_groups = [
+            issue
+            for issue in self._state_data["task"]["hidden_issues"]
+            if issue["type"] == "duplicate" and self._is_issue_unresolved(issue, self._state_data["table"])
+        ]
+        if not duplicate_groups:
+            result["unnecessary_action"] = True
+            result["error_type"] = "no_duplicate_available"
+            return
+
+        candidate_rows = set(duplicate_groups[0].get("rows", []))
+        target_row_id = action.row_id or max(candidate_rows)
+
+        if target_row_id not in candidate_rows:
+            result["unnecessary_action"] = True
+            result["error_type"] = "invalid_duplicate_target"
+            return
+
+        removed = self._remove_row_by_id(target_row_id)
+        if not removed:
+            result["unnecessary_action"] = True
+            result["error_type"] = "missing_row"
+
+    def _delete_row(self, action: Action, result: MutableMapping[str, Any]) -> None:
+        """Delete a row and mark destructive behavior when the target is unsafe."""
+
+        target_row = self._get_row_by_id(action.row_id)
+        if target_row is None:
+            result["unnecessary_action"] = True
+            result["error_type"] = "missing_row"
+            return
+
+        if self._row_is_protected(action.row_id):
+            result["wrong_deletion"] = True
+            result["destructive_action"] = True
+            result["error_type"] = "protected_row"
+            result["mistake_keys"].append(f"{action.action_type}:protected_row")
+        elif not self._row_belongs_to_removable_issue(action.row_id):
+            result["wrong_deletion"] = True
+            result["destructive_action"] = True
+            result["error_type"] = "wrong_deletion"
+            result["mistake_keys"].append(f"{action.action_type}:wrong_deletion")
+
+        self._remove_row_by_id(action.row_id)
+
+    def _fill_missing(self, action: Action, result: MutableMapping[str, Any]) -> None:
+        """Fill a missing field on the target row or the first matching missing cell."""
+
+        target_row = self._resolve_missing_target_row(action.row_id, action.column)
+        if target_row is None or action.column is None:
+            result["unnecessary_action"] = True
+            result["error_type"] = "missing_target"
+            return
+
+        if not self._is_missing_value(target_row.get(action.column)):
+            result["unnecessary_action"] = True
+            result["error_type"] = "cell_not_missing"
+            return
+
+        target_row[action.column] = action.value
+
+    def _normalize_column(self, action: Action, result: MutableMapping[str, Any]) -> None:
+        """Normalize a supported column using deterministic, minimal edits."""
+
+        if action.column is None:
+            result["unnecessary_action"] = True
+            result["error_type"] = "missing_column"
+            return
+
+        changed_rows = 0
+        for row in self._state_data["table"]:
+            original = row.get(action.column)
+            normalized = self._normalized_value(action.column, original)
+            if normalized is None or normalized == original:
+                continue
+
+            # Keep trap rows stable unless the value is actually invalid.
+            if self._row_is_protected(row.get("row_id")) and self._value_is_valid(
+                action.column, original
+            ):
+                continue
+
+            row[action.column] = normalized
+            changed_rows += 1
+
+        if changed_rows == 0:
+            result["unnecessary_action"] = True
+            result["error_type"] = "no_normalization_needed"
+
+    def _populate_result_signals(
+        self,
+        action: Action,
+        table_before: List[Dict[str, Any]],
+        table_after: List[Dict[str, Any]],
+        issues_before: List[str],
+        issues_after: List[str],
+        result: MutableMapping[str, Any],
+    ) -> None:
+        """Derive reward signals from before/after state transitions."""
+
+        task_definition: TaskDefinition = self._state_data["task"]
+        hidden_before = self._issue_type_counts(table_before, task_definition)
+        hidden_after = self._issue_type_counts(table_after, task_definition)
+
+        if hidden_after.get("duplicate", 0) < hidden_before.get("duplicate", 0):
+            result["correct_duplicate_removal"] = True
+
+        if hidden_after.get("missing_value", 0) < hidden_before.get("missing_value", 0):
+            result["fixed_missing_value"] = True
+
+        normalization_before = hidden_before.get("inconsistent_casing", 0) + hidden_before.get(
+            "invalid_format", 0
+        )
+        normalization_after = hidden_after.get("inconsistent_casing", 0) + hidden_after.get(
+            "invalid_format", 0
+        )
+        if (
+            action.action_type == "normalize_column"
+            and normalization_after < normalization_before
+        ):
+            result["correct_normalization"] = True
+
+        if action.action_type == "validate" and not issues_after:
+            result["validation_success"] = True
+            result["task_completed"] = True
+
+        if not issues_after:
+            result["task_completed"] = True
+
+        issue_delta = max(0, len(issues_before) - len(issues_after))
+        result["progress_delta"] = round(
+            issue_delta / float(self._state_data["initial_issue_count"]),
+            4,
+        )
+
+        if issue_delta > 0 and any(self._state_data["mistakes"].values()):
+            result["corrected_previous_mistake"] = True
+
+        if action.action_type == "noop" and issues_after:
+            result["unnecessary_action"] = True
+            result["error_type"] = result.get("error_type", "noop")
+
+    def _build_observation(self) -> Observation:
+        """Construct the typed observation returned to callers."""
+
+        task_definition: TaskDefinition = self._state_data["task"]
+        issue_messages = self._current_issue_messages(self._state_data["table"], task_definition)
+        progress = self._compute_progress(issue_messages)
+        return Observation(
+            goal=task_definition["goal"],
+            table=deepcopy(self._state_data["table"]),
+            issues=issue_messages,
+            history=list(self._state_data["history"]),
+            mistakes=deepcopy(self._state_data["mistakes"]),
+            hints=list(self._state_data["hints"]),
+            progress=progress,
+            steps_remaining=int(self._state_data["steps_remaining"]),
+        )
+
+    def _compute_progress(self, issue_messages: List[str]) -> float:
+        """Estimate progress from the current unresolved issue count."""
+
+        baseline = float(self._state_data["initial_issue_count"])
+        remaining = min(len(issue_messages), self._state_data["initial_issue_count"])
+        resolved_fraction = 1.0 - (remaining / baseline)
+        return round(max(0.0, min(1.0, resolved_fraction)), 4)
+
+    def _current_issue_messages(
+        self, table: List[Dict[str, Any]], task_definition: TaskDefinition
+    ) -> List[str]:
+        """Return unresolved issue descriptions plus validation-rule failures."""
+
+        messages: List[str] = []
+        for issue in task_definition["hidden_issues"]:
+            if self._is_issue_unresolved(issue, table):
+                description = issue.get("description")
+                if description:
+                    messages.append(description)
+
+        messages.extend(self._validation_failures(table, task_definition))
+        return messages
+
+    def _validation_failures(
+        self, table: List[Dict[str, Any]], task_definition: TaskDefinition
+    ) -> List[str]:
+        """Evaluate rule-based outcome constraints beyond the hidden issue list."""
+
+        return task_failure_messages(task_definition, table, self._state_data)
+
+    def _issue_type_counts(
+        self, table: List[Dict[str, Any]], task_definition: TaskDefinition
+    ) -> Dict[str, int]:
+        """Count unresolved hidden issues by type."""
+
+        counts: Dict[str, int] = {}
+        for issue in task_definition["hidden_issues"]:
+            if self._is_issue_unresolved(issue, table):
+                issue_type = issue["type"]
+                counts[issue_type] = counts.get(issue_type, 0) + 1
+        return counts
+
+    def _is_issue_unresolved(self, issue: HiddenIssue, table: List[Dict[str, Any]]) -> bool:
+        """Determine whether a hidden issue is still unresolved."""
+
+        issue_type = issue["type"]
+        table_by_row_id = self._table_by_row_id(table)
+
+        if issue_type == "valid_trap":
+            return False
+
+        if issue_type in {"duplicate", "conflict"}:
+            rows = issue.get("rows", [])
+            return all(row_id in table_by_row_id for row_id in rows)
+
+        if issue_type == "missing_value":
+            row = table_by_row_id.get(issue.get("row"))
+            column = issue.get("column")
+            return row is not None and column is not None and self._is_missing_value(row.get(column))
+
+        if issue_type == "inconsistent_casing":
+            column = issue.get("column")
+            return any(
+                row_id in table_by_row_id
+                and self._needs_title_case(str(table_by_row_id[row_id].get(column, "")))
+                for row_id in issue.get("rows", [])
+            )
+
+        if issue_type == "invalid_format":
+            row = table_by_row_id.get(issue.get("row"))
+            column = issue.get("column")
+            return row is not None and column is not None and not self._value_is_valid(
+                column, row.get(column)
+            )
+
+        if issue_type == "constraint_violation" and issue.get("constraint") == "unique_email":
+            rows = issue.get("rows", [])
+            emails = [
+                table_by_row_id[row_id].get("email")
+                for row_id in rows
+                if row_id in table_by_row_id
+            ]
+            return len(emails) != len(set(emails))
+
+        return False
+
+    def _update_hints(self, result: Mapping[str, Any], issues_after: List[str]) -> None:
+        """Add deterministic hints when the agent stalls or accumulates mistakes."""
+
+        if not issues_after:
+            return
+
+        global_wrong_deletion_count = sum(
+            count
+            for key, count in self._global_mistake_memory.items()
+            if key == "wrong_deletion" or key.endswith(":wrong_deletion")
+        )
+        if global_wrong_deletion_count >= 3:
+            hint = (
+                "You are repeatedly deleting valid rows. Try resolving issues "
+                "instead of deleting."
+            )
+            if hint not in self._state_data["hints"]:
+                self._state_data["hints"].append(hint)
+
+        total_mistakes = sum(self._state_data["mistakes"].values())
+        should_hint = bool(result.get("unnecessary_action")) or bool(
+            result.get("wrong_deletion")
+        ) or total_mistakes >= 2 or float(result.get("progress_delta", 0.0)) == 0.0
+
+        if not should_hint:
+            return
+
+        next_hint = self._build_hint(issues_after[0])
+        if next_hint not in self._state_data["hints"]:
+            self._state_data["hints"].append(next_hint)
+
+    def _build_hint(self, issue_message: str) -> str:
+        """Map unresolved issue descriptions to small, actionable hints."""
+
+        lowered = issue_message.lower()
+        if "duplicate" in lowered:
+            return "Look for rows that describe the same entity and keep only one representative record."
+        if "missing" in lowered:
+            return "A required field is still empty. Fill the missing value instead of deleting the row."
+        if "email" in lowered and "format" in lowered:
+            return "Normalize only the invalid email values; valid addresses should be preserved."
+        if "phone" in lowered:
+            return "Repair only phone values that are actually malformed."
+        if "title-case" in lowered or "casing" in lowered:
+            return "Normalize text columns to a consistent title-case style."
+        if "unchanged" in lowered:
+            return "Some unusual-looking rows are valid traps and should be preserved."
+        return "Focus on the first unresolved issue and prefer minimal corrective actions."
+
+    def _record_mistake_memory(
+        self, action: Action, result: Mapping[str, Any]
+    ) -> None:
+        """Persist mistake events so hinting can look at prior failures."""
+
+        for key, count in self._state_data["mistakes"].items():
+            if count <= 0:
+                continue
+            if action.action_id:
+                memory_entry = f"{action.action_id}:{key}:{count}"
+            else:
+                memory_entry = f"{action.action_type}:{key}:{count}"
+            if memory_entry not in self._state_data["mistake_memory"]:
+                self._state_data["mistake_memory"].append(memory_entry)
+
+            self._global_mistake_memory[key] = (
+                self._global_mistake_memory.get(key, 0) + 1
+            )
+            category_key = key.split(":")[-1]
+            self._global_mistake_memory[category_key] = (
+                self._global_mistake_memory.get(category_key, 0) + 1
+            )
+
+        if result.get("destructive_action"):
+            entry = f"{action.action_type}:destructive_action"
+            if entry not in self._state_data["mistake_memory"]:
+                self._state_data["mistake_memory"].append(entry)
+
+    def _resolve_missing_target_row(
+        self, row_id: Optional[int], column: Optional[str]
+    ) -> Optional[Dict[str, Any]]:
+        """Choose the requested row or the first matching missing-value row."""
+
+        if row_id is not None:
+            return self._get_row_by_id(row_id)
+
+        if column is None:
+            return None
+
+        for row in self._state_data["table"]:
+            if self._is_missing_value(row.get(column)):
+                return row
+        return None
+
+    def _normalized_value(self, column: str, value: Any) -> Any:
+        """Return a normalized value for supported columns."""
+
+        if not isinstance(value, str):
+            return value
+
+        if column in {"name", "city"}:
+            return value.title()
+
+        if column == "email" and not self._is_valid_email(value):
+            normalized = value.strip().lower()
+            normalized = normalized.replace("[at]", "@").replace(" at ", "@")
+            if "@" not in normalized and normalized.endswith(".example.com"):
+                normalized = normalized.replace(".example.com", "@example.com", 1)
+            if "@" in normalized and "." not in normalized.split("@", 1)[1]:
+                normalized = normalized + ".com"
+            return normalized
+
+        if column == "phone" and not self._is_valid_phone(value):
+            digits = re.sub(r"\D", "", value)
+            if len(digits) == 11 and digits.startswith("1"):
+                digits = digits[1:]
+            if len(digits) == 10:
+                return f"{digits[0:3]}-{digits[3:6]}-{digits[6:10]}"
+        return value
+
+    def _value_is_valid(self, column: str, value: Any) -> bool:
+        """Validate known column types used by the tasks."""
+
+        if value is None:
+            return False
+        if column == "email":
+            return self._is_valid_email(str(value))
+        if column == "phone":
+            return self._is_valid_phone(str(value))
+        if column in {"name", "city"}:
+            return not self._needs_title_case(str(value))
+        return True
+
+    def _is_valid_email(self, value: str) -> bool:
+        """Return whether the supplied email string looks valid."""
+
+        return bool(EMAIL_PATTERN.match(value.strip()))
+
+    def _is_valid_phone(self, value: str) -> bool:
+        """Return whether the supplied phone value is valid for this environment."""
+
+        digits = re.sub(r"\D", "", value)
+        return len(digits) == 10 or (len(digits) == 11 and digits.startswith("1"))
+
+    def _needs_title_case(self, value: str) -> bool:
+        """Detect whether a string still needs title-case normalization."""
+
+        cleaned = value.strip()
+        return bool(cleaned) and cleaned != cleaned.title()
+
+    def _has_missing_required_values(
+        self, table: Iterable[Dict[str, Any]], required_columns: Iterable[str]
+    ) -> bool:
+        """Check whether any required field remains missing."""
+
+        for row in table:
+            for column in required_columns:
+                if self._is_missing_value(row.get(column)):
+                    return True
+        return False
+
+    def _has_duplicates(self, table: Iterable[Dict[str, Any]], column: str) -> bool:
+        """Check whether a column contains duplicate non-empty values."""
+
+        values = [row.get(column) for row in table if row.get(column) not in (None, "")]
+        return len(values) != len(set(values))
+
+    def _column_has_invalid_email(
+        self, table: Iterable[Dict[str, Any]], column: str
+    ) -> bool:
+        """Check whether any remaining email value is invalid."""
+
+        return any(
+            row.get(column) not in (None, "") and not self._is_valid_email(str(row.get(column)))
+            for row in table
+        )
+
+    def _column_has_invalid_phone(
+        self, table: Iterable[Dict[str, Any]], column: str
+    ) -> bool:
+        """Check whether any remaining phone value is invalid."""
+
+        return any(
+            row.get(column) not in (None, "") and not self._is_valid_phone(str(row.get(column)))
+            for row in table
+        )
+
+    def _column_needs_title_case(
+        self, table: Iterable[Dict[str, Any]], column: str
+    ) -> bool:
+        """Check whether any remaining value still violates title-case normalization."""
+
+        return any(
+            isinstance(row.get(column), str) and self._needs_title_case(str(row.get(column)))
+            for row in table
+        )
+
+    def _row_has_changed_from_initial(
+        self, row_id: int, current_table: List[Dict[str, Any]]
+    ) -> bool:
+        """Check whether a protected row has changed relative to the task start."""
+
+        current_row = self._table_by_row_id(current_table).get(row_id)
+        initial_row = self._state_data["initial_table_by_row_id"].get(row_id)
+        if current_row is None or initial_row is None:
+            return True
+        return current_row != initial_row
+
+    def _row_is_protected(self, row_id: Optional[int]) -> bool:
+        """Return whether a row is marked as a valid trap in the current task."""
+
+        if row_id is None:
+            return False
+        for issue in self._state_data["task"]["hidden_issues"]:
+            if issue["type"] == "valid_trap" and issue.get("row") == row_id:
+                return True
+        return False
+
+    def _row_belongs_to_removable_issue(self, row_id: Optional[int]) -> bool:
+        """Return whether deleting a row could plausibly resolve a structural issue."""
+
+        if row_id is None:
+            return False
+        for issue in self._state_data["task"]["hidden_issues"]:
+            if issue["type"] in {"duplicate", "conflict", "constraint_violation"} and row_id in issue.get(
+                "rows", []
+            ):
+                return True
+        return False
+
+    def _remove_row_by_id(self, row_id: Optional[int]) -> bool:
+        """Remove a row by id and report whether a row was deleted."""
+
+        if row_id is None:
+            return False
+        table = self._state_data["table"]
+        for index, row in enumerate(table):
+            if row.get("row_id") == row_id:
+                del table[index]
+                return True
+        return False
+
+    def _get_row_by_id(self, row_id: Optional[int]) -> Optional[Dict[str, Any]]:
+        """Return a mutable row reference by id."""
+
+        if row_id is None:
+            return None
+        for row in self._state_data["table"]:
+            if row.get("row_id") == row_id:
+                return row
+        return None
+
+    def _table_by_row_id(self, table: List[Dict[str, Any]]) -> Dict[int, Dict[str, Any]]:
+        """Index a table by row id."""
+
+        return {
+            int(row["row_id"]): deepcopy(row)
+            for row in table
+            if row.get("row_id") is not None
+        }
+
+    def _is_missing_value(self, value: Any) -> bool:
+        """Return whether a cell should be treated as missing."""
+
+        return value is None or value == ""
+
+    def _format_history(self, action: Action) -> str:
+        """Return a compact history entry for the applied action."""
+
+        details = []
+        if action.row_id is not None:
+            details.append(f"row_id={action.row_id}")
+        if action.column is not None:
+            details.append(f"column={action.column}")
+        if action.value is not None:
+            details.append(f"value={action.value}")
+        detail_text = ", ".join(details)
+        return f"{action.action_type}({detail_text})" if detail_text else action.action_type
+
+
+class DataOpsGymEnv(DataOpsEnv):
+    """Compatibility wrapper matching the configured OpenEnv entrypoint."""
+
+    pass
+
+__all__ = ["DataOpsEnv", "DataOpsGymEnv"]

grader.py

@@ -0,0 +1,592 @@
+"""Evaluation and grading interfaces for ``dataops-gym``.
+
+This module is responsible for validating outputs, scoring task results, and
+capturing assessment metadata independently from task execution logic.
+"""
+
+from __future__ import annotations
+
+import re
+from typing import Any, Dict, Iterable, Mapping, MutableMapping, Optional, Tuple
+
+
+# Dense reward values are intentionally small and additive so the agent receives
+# feedback for intermediate progress without requiring full task completion.
+CORRECT_DUPLICATE_REMOVAL_REWARD = 0.3
+CORRECT_NORMALIZATION_REWARD = 0.2
+FIX_MISSING_VALUE_REWARD = 0.2
+VALIDATION_SUCCESS_REWARD = 0.2
+EFFICIENCY_BONUS = 0.2
+RECOVERY_BONUS = 0.25
+STEP_PENALTY = -0.02
+PROGRESS_REWARD_SCALE = 0.3
+
+# Penalties are split into:
+# 1. a direct penalty for the current bad action, and
+# 2. an escalating repetition penalty if the same mistake keeps happening.
+WRONG_DELETION_PENALTY = -0.3
+UNNECESSARY_ACTION_PENALTY = -0.1
+NOOP_PENALTY = -0.05
+DESTRUCTIVE_ACTION_PENALTY = -0.4
+
+FIRST_REPEAT_PENALTY = -0.1
+SECOND_REPEAT_PENALTY = -0.2
+THIRD_OR_MORE_REPEAT_PENALTY = -0.4
+EMAIL_PATTERN = re.compile(r"^[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}$")
+
+
+def detect_repeated_mistake(mistakes: Mapping[str, int], mistake_key: str) -> int:
+    """Return how many times a mistake has already occurred before this step."""
+
+    return int(mistakes.get(mistake_key, 0))
+
+
+def track_mistake(state: MutableMapping[str, Any], mistake_key: str) -> int:
+    """Update the mistake counter in state and return the new occurrence count."""
+
+    mistakes = state.setdefault("mistakes", {})
+    if not isinstance(mistakes, dict):
+        raise ValueError("state['mistakes'] must be a dictionary for mistake tracking")
+
+    current_count = int(mistakes.get(mistake_key, 0))
+    new_count = current_count + 1
+    mistakes[mistake_key] = new_count
+    return new_count
+
+
+def repeated_mistake_penalty(occurrence_count: int) -> float:
+    """Return the escalating penalty for repeated mistakes."""
+
+    if occurrence_count <= 1:
+        return FIRST_REPEAT_PENALTY
+    if occurrence_count == 2:
+        return SECOND_REPEAT_PENALTY
+    return THIRD_OR_MORE_REPEAT_PENALTY
+
+
+def _to_bool(mapping: Mapping[str, Any], key: str) -> bool:
+    """Normalize truthy result flags into deterministic boolean checks."""
+
+    return bool(mapping.get(key, False))
+
+
+def _mistake_key(
+    action: Mapping[str, Any],
+    result: Mapping[str, Any],
+    fallback_key: str,
+) -> str:
+    """Build an action-specific mistake key with a safe fallback."""
+
+    action_type = action.get("action_type")
+    error_type = result.get("error_type", "general")
+
+    if action_type:
+        return f"{action_type}:{error_type}"
+    return fallback_key
+
+
+def _clamp_reward(value: float) -> float:
+    """Keep rewards in the required [-1.0, 1.0] range."""
+
+    return max(-1.0, min(1.0, round(value, 4)))
+
+
+def _clamp_score(value: float) -> float:
+    """Keep task-level scores in the required [0.0, 1.0] range."""
+
+    return max(0.0, min(1.0, round(value, 4)))
+
+
+def _is_missing_value(value: Any) -> bool:
+    """Return whether a cell should be considered missing."""
+
+    return value is None or value == ""
+
+
+def _is_valid_email(value: str) -> bool:
+    """Validate email formatting used by task graders."""
+
+    return bool(EMAIL_PATTERN.match(value.strip()))
+
+
+def _is_valid_phone(value: str) -> bool:
+    """Validate phone formatting used by task graders."""
+
+    digits = re.sub(r"\D", "", value)
+    return len(digits) == 10 or (len(digits) == 11 and digits.startswith("1"))
+
+
+def _needs_title_case(value: str) -> bool:
+    """Return whether text still violates title-case normalization."""
+
+    cleaned = value.strip()
+    return bool(cleaned) and cleaned != cleaned.title()
+
+
+def _has_duplicates(table: Iterable[Dict[str, Any]], column: str) -> bool:
+    """Check whether a column contains duplicate non-empty values."""
+
+    values = [row.get(column) for row in table if row.get(column) not in (None, "")]
+    return len(values) != len(set(values))
+
+
+def _table_by_row_id(table: Iterable[Dict[str, Any]]) -> Dict[int, Dict[str, Any]]:
+    """Index a table by ``row_id`` for deterministic issue evaluation."""
+
+    return {
+        int(row["row_id"]): dict(row)
+        for row in table
+        if row.get("row_id") is not None
+    }
+
+
+def _is_issue_resolved(issue: Mapping[str, Any], table_by_row_id: Dict[int, Dict[str, Any]]) -> bool:
+    """Return whether a structured hidden issue has been resolved."""
+
+    issue_type = issue.get("type")
+
+    if issue_type == "valid_trap":
+        return True
+
+    if issue_type in {"duplicate", "conflict"}:
+        rows = issue.get("rows", [])
+        return not all(row_id in table_by_row_id for row_id in rows)
+
+    if issue_type == "missing_value":
+        row = table_by_row_id.get(issue.get("row"))
+        column = issue.get("column")
+        return row is None or column is None or not _is_missing_value(row.get(column))
+
+    if issue_type == "inconsistent_casing":
+        column = issue.get("column")
+        rows = issue.get("rows", [])
+        return not any(
+            row_id in table_by_row_id
+            and isinstance(table_by_row_id[row_id].get(column), str)
+            and _needs_title_case(str(table_by_row_id[row_id].get(column)))
+            for row_id in rows
+        )
+
+    if issue_type == "invalid_format":
+        row = table_by_row_id.get(issue.get("row"))
+        column = issue.get("column")
+        if row is None or column is None:
+            return True
+        value = row.get(column)
+        if column == "email":
+            return _is_valid_email(str(value))
+        if column == "phone":
+            return _is_valid_phone(str(value))
+        return True
+
+    if issue_type == "constraint_violation" and issue.get("constraint") == "unique_email":
+        rows = issue.get("rows", [])
+        emails = [
+            table_by_row_id[row_id].get("email")
+            for row_id in rows
+            if row_id in table_by_row_id
+        ]
+        return len(emails) == len(set(emails))
+
+    return True
+
+
+def _task_check_results(
+    task_definition: Mapping[str, Any],
+    table: Iterable[Dict[str, Any]],
+    state: Optional[Mapping[str, Any]] = None,
+) -> list[Dict[str, Any]]:
+    """Build explicit pass/fail checks for final grading and validation."""
+
+    rows = [dict(row) for row in table]
+    table_by_row_id = _table_by_row_id(rows)
+    expected_outcome = dict(task_definition.get("expected_outcome", {}))
+    checks: list[Dict[str, Any]] = []
+
+    expected_row_count = expected_outcome.get("expected_row_count")
+    if expected_row_count is not None:
+        checks.append(
+            {
+                "name": "expected_row_count",
+                "passed": len(rows) == expected_row_count,
+                "message": f"Expected exactly {expected_row_count} rows in the cleaned table.",
+            }
+        )
+
+    expected_row_range = expected_outcome.get("expected_row_count_range")
+    if expected_row_range is not None:
+        checks.append(
+            {
+                "name": "expected_row_count_range",
+                "passed": expected_row_range["min"] <= len(rows) <= expected_row_range["max"],
+                "message": (
+                    "Expected the cleaned table to contain between "
+                    f"{expected_row_range['min']} and {expected_row_range['max']} rows."
+                ),
+            }
+        )
+
+    required_columns = expected_outcome.get(
+        "required_non_null_columns", task_definition.get("required_columns", [])
+    )
+    if required_columns:
+        checks.append(
+            {
+                "name": "required_non_null_columns",
+                "passed": not any(
+                    _is_missing_value(row.get(column))
+                    for row in rows
+                    for column in required_columns
+                ),
+                "message": "Required columns must be populated for all remaining rows.",
+            }
+        )
+
+    for unique_column in expected_outcome.get("unique_by", []):
+        checks.append(
+            {
+                "name": f"unique_by:{unique_column}",
+                "passed": not _has_duplicates(rows, unique_column),
+                "message": f"Values in '{unique_column}' must remain unique.",
+            }
+        )
+
+    for column, rule in expected_outcome.get("normalized_columns", {}).items():
+        if rule == "title_case":
+            checks.append(
+                {
+                    "name": f"normalized_column:{column}",
+                    "passed": not any(
+                        isinstance(row.get(column), str)
+                        and _needs_title_case(str(row.get(column)))
+                        for row in rows
+                    ),
+                    "message": f"Column '{column}' should use a consistent title-case style.",
+                }
+            )
+
+    for column, rule in expected_outcome.get("format_rules", {}).items():
+        if rule == "valid_email":
+            checks.append(
+                {
+                    "name": f"valid_email:{column}",
+                    "passed": not any(
+                        row.get(column) not in (None, "")
+                        and not _is_valid_email(str(row.get(column)))
+                        for row in rows
+                    ),
+                    "message": "All remaining email values must use a valid email format.",
+                }
+            )
+        if rule == "normalized_phone":
+            checks.append(
+                {
+                    "name": f"normalized_phone:{column}",
+                    "passed": not any(
+                        row.get(column) not in (None, "")
+                        and not _is_valid_phone(str(row.get(column)))
+                        for row in rows
+                    ),
+                    "message": "All remaining phone values must use a consistent valid format.",
+                }
+            )
+
+    initial_rows = {}
+    if state is not None:
+        initial_rows = dict(state.get("initial_table_by_row_id", {}))
+
+    for row_id in expected_outcome.get("must_preserve_valid_rows", []):
+        current_row = table_by_row_id.get(row_id)
+        checks.append(
+            {
+                "name": f"preserve_valid_row:{row_id}",
+                "passed": current_row is not None and current_row == initial_rows.get(row_id),
+                "message": f"Valid row {row_id} should remain logically unchanged.",
+            }
+        )
+
+    for row_group in expected_outcome.get("exactly_one_of_rows", []):
+        surviving = [row_id for row_id in row_group if row_id in table_by_row_id]
+        checks.append(
+            {
+                "name": f"exactly_one_of_rows:{','.join(str(row_id) for row_id in row_group)}",
+                "passed": len(surviving) == 1,
+                "message": f"Exactly one of rows {row_group} should remain in the cleaned table.",
+            }
+        )
+
+    for row_id in expected_outcome.get("rows_must_survive", []):
+        checks.append(
+            {
+                "name": f"rows_must_survive:{row_id}",
+                "passed": row_id in table_by_row_id,
+                "message": f"Row {row_id} must still be present in the cleaned table.",
+            }
+        )
+
+    for row_id in expected_outcome.get("rows_must_be_removed", []):
+        checks.append(
+            {
+                "name": f"rows_must_be_removed:{row_id}",
+                "passed": row_id not in table_by_row_id,
+                "message": f"Row {row_id} should not remain in the cleaned table.",
+            }
+        )
+
+    for issue in task_definition.get("hidden_issues", []):
+        if issue.get("type") == "valid_trap":
+            continue
+        message = issue.get("description") or f"Issue '{issue.get('type')}' must be resolved."
+        checks.append(
+            {
+                "name": f"hidden_issue:{issue.get('type')}",
+                "passed": _is_issue_resolved(issue, table_by_row_id),
+                "message": message,
+            }
+        )
+
+    return checks
+
+
+def _calculate_reward(
+    state: MutableMapping[str, Any],
+    action: Mapping[str, Any],
+    result: MutableMapping[str, Any],
+) -> float:
+    """Compute the deterministic scalar reward for a single environment step."""
+
+    reward = 0.0
+
+    # Every step incurs a small cost so the agent is encouraged to solve the
+    # task quickly instead of exploring indefinitely.
+    reward += STEP_PENALTY
+
+    # Intermediate rewards encourage the agent to make progress even when the
+    # dataset is not fully clean yet.
+    if _to_bool(result, "correct_duplicate_removal"):
+        reward += CORRECT_DUPLICATE_REMOVAL_REWARD
+
+    if _to_bool(result, "correct_normalization"):
+        reward += CORRECT_NORMALIZATION_REWARD
+
+    if _to_bool(result, "fixed_missing_value") or _to_bool(
+        result, "fixing_missing_values"
+    ):
+        reward += FIX_MISSING_VALUE_REWARD
+
+    if _to_bool(result, "validation_success"):
+        reward += VALIDATION_SUCCESS_REWARD
+
+    if _to_bool(result, "corrected_previous_mistake"):
+        reward += RECOVERY_BONUS
+
+    if _to_bool(result, "noop"):
+        reward += NOOP_PENALTY
+
+    if _to_bool(result, "destructive_action"):
+        reward += DESTRUCTIVE_ACTION_PENALTY
+
+    # Progress-based shaping provides a smoother learning signal for partial
+    # improvement, even when a step does not fully resolve a visible issue.
+    progress_delta = float(result.get("progress_delta", 0.0))
+    progress_delta = max(0.0, min(1.0, progress_delta))
+    reward += progress_delta * PROGRESS_REWARD_SCALE
+
+    # Explicitly penalize steps that fail to improve task progress so agents do
+    # not learn that random but harmless actions are equivalent to useful ones.
+    if progress_delta == 0.0:
+        reward -= 0.05
+
+    # Direct penalties handle obviously harmful moves. Repetition is tracked
+    # separately so the same bad behavior becomes more expensive over time.
+    if _to_bool(result, "wrong_deletion"):
+        reward += WRONG_DELETION_PENALTY
+        mistake_key = _mistake_key(action, result, "wrong_deletion")
+        occurrence_count = track_mistake(state, mistake_key)
+        reward += repeated_mistake_penalty(occurrence_count)
+
+    if _to_bool(result, "unnecessary_action"):
+        reward += UNNECESSARY_ACTION_PENALTY
+        mistake_key = _mistake_key(action, result, "unnecessary_action")
+        occurrence_count = track_mistake(state, mistake_key)
+        reward += repeated_mistake_penalty(occurrence_count)
+
+    # Support arbitrary custom mistake keys in addition to the built-in ones.
+    for mistake_key in result.get("mistake_keys", []):
+        if mistake_key not in {"wrong_deletion", "unnecessary_action"}:
+            occurrence_count = track_mistake(state, str(mistake_key))
+            reward += repeated_mistake_penalty(occurrence_count)
+
+    # Reward early completion only when the task finishes with steps still
+    # available. This creates a simple deterministic efficiency incentive.
+    if _to_bool(result, "task_completed") and int(state.get("steps_remaining", 0)) > 0:
+        reward += EFFICIENCY_BONUS
+
+    return _clamp_reward(reward)
+
+
+def grade_step(
+    state: MutableMapping[str, Any],
+    action: Mapping[str, Any],
+    result: MutableMapping[str, Any],
+) -> float:
+    """Compute a deterministic dense reward for a single environment step."""
+
+    return _calculate_reward(state, action, result)
+
+
+def grade_step_details(
+    state: MutableMapping[str, Any],
+    action: Mapping[str, Any],
+    result: MutableMapping[str, Any],
+) -> Tuple[float, Dict[str, Any]]:
+    """Compute reward plus a structured component breakdown for debugging."""
+
+    previous_mistakes = {
+        key: int(value)
+        for key, value in state.get("mistakes", {}).items()
+    }
+    reward = grade_step(state, action, result)
+
+    wrong_deletion_repeat_penalty = 0.0
+    if result.get("wrong_deletion"):
+        mistake_key = _mistake_key(action, result, "wrong_deletion")
+        occurrence_count = int(state.get("mistakes", {}).get(mistake_key, 0))
+        if occurrence_count > int(previous_mistakes.get(mistake_key, 0)):
+            wrong_deletion_repeat_penalty = repeated_mistake_penalty(occurrence_count)
+
+    unnecessary_repeat_penalty = 0.0
+    if result.get("unnecessary_action"):
+        mistake_key = _mistake_key(action, result, "unnecessary_action")
+        occurrence_count = int(state.get("mistakes", {}).get(mistake_key, 0))
+        if occurrence_count > int(previous_mistakes.get(mistake_key, 0)):
+            unnecessary_repeat_penalty = repeated_mistake_penalty(occurrence_count)
+
+    components: Dict[str, Any] = {
+        "step_penalty": STEP_PENALTY,
+        "duplicate_reward": (
+            CORRECT_DUPLICATE_REMOVAL_REWARD
+            if result.get("correct_duplicate_removal")
+            else 0.0
+        ),
+        "normalization_reward": (
+            CORRECT_NORMALIZATION_REWARD
+            if result.get("correct_normalization")
+            else 0.0
+        ),
+        "missing_value_reward": (
+            FIX_MISSING_VALUE_REWARD if result.get("fixed_missing_value") else 0.0
+        ),
+        "validation_reward": (
+            VALIDATION_SUCCESS_REWARD if result.get("validation_success") else 0.0
+        ),
+        "penalties": {
+            "wrong_deletion": (
+                WRONG_DELETION_PENALTY if result.get("wrong_deletion") else 0.0
+            ),
+            "unnecessary_action": (
+                UNNECESSARY_ACTION_PENALTY if result.get("unnecessary_action") else 0.0
+            ),
+            "wrong_deletion_repeat": wrong_deletion_repeat_penalty,
+            "unnecessary_action_repeat": unnecessary_repeat_penalty,
+            "noop": NOOP_PENALTY if result.get("noop") else 0.0,
+            "destructive_action": (
+                DESTRUCTIVE_ACTION_PENALTY
+                if result.get("destructive_action")
+                else 0.0
+            ),
+        },
+        "progress_reward": round(
+            max(0.0, min(1.0, float(result.get("progress_delta", 0.0))))
+            * PROGRESS_REWARD_SCALE,
+            4,
+        ),
+        "recovery_bonus": (
+            RECOVERY_BONUS if result.get("corrected_previous_mistake") else 0.0
+        ),
+        "efficiency_bonus": (
+            EFFICIENCY_BONUS
+            if result.get("task_completed") and int(state.get("steps_remaining", 0)) > 0
+            else 0.0
+        ),
+    }
+
+    if float(result.get("progress_delta", 0.0)) == 0.0:
+        components["no_progress_penalty"] = -0.05
+
+    result["reward_components"] = components
+    result["reward_total"] = reward
+    return reward, components
+
+
+def grade_task_result(
+    task_definition: Mapping[str, Any],
+    table: Iterable[Dict[str, Any]],
+    state: Optional[Mapping[str, Any]] = None,
+) -> float:
+    """Compute a deterministic final task score between 0.0 and 1.0."""
+
+    checks = _task_check_results(task_definition, table, state)
+    if not checks:
+        return 0.0
+    return _clamp_score(
+        sum(1.0 for check in checks if check["passed"]) / len(checks)
+    )
+
+
+def task_failure_messages(
+    task_definition: Mapping[str, Any],
+    table: Iterable[Dict[str, Any]],
+    state: Optional[Mapping[str, Any]] = None,
+) -> list[str]:
+    """Return explicit failure messages for unresolved outcome checks."""
+
+    return [
+        str(check["message"])
+        for check in _task_check_results(task_definition, table, state)
+        if not bool(check["passed"])
+    ]
+
+
+def grade_easy_cleaning_task(
+    task_definition: Mapping[str, Any],
+    table: Iterable[Dict[str, Any]],
+    state: Optional[Mapping[str, Any]] = None,
+) -> float:
+    """Grade the easy cleaning task on a 0.0–1.0 scale."""
+
+    return grade_task_result(task_definition, table, state)
+
+
+def grade_medium_normalization_task(
+    task_definition: Mapping[str, Any],
+    table: Iterable[Dict[str, Any]],
+    state: Optional[Mapping[str, Any]] = None,
+) -> float:
+    """Grade the medium normalization task on a 0.0–1.0 scale."""
+
+    return grade_task_result(task_definition, table, state)
+
+
+def grade_hard_conflict_resolution_task(
+    task_definition: Mapping[str, Any],
+    table: Iterable[Dict[str, Any]],
+    state: Optional[Mapping[str, Any]] = None,
+) -> float:
+    """Grade the hard conflict-resolution task on a 0.0–1.0 scale."""
+
+    return grade_task_result(task_definition, table, state)
+
+
+__all__ = [
+    "detect_repeated_mistake",
+    "grade_step",
+    "grade_step_details",
+    "grade_task_result",
+    "task_failure_messages",
+    "grade_easy_cleaning_task",
+    "grade_medium_normalization_task",
+    "grade_hard_conflict_resolution_task",
+    "repeated_mistake_penalty",
+    "track_mistake",
+]

inference.py

@@ -0,0 +1,989 @@
+"""Inference entrypoints for ``dataops-gym``.
+
+This runner keeps the hackathon-required OpenAI-compatible model interface, but
+adds a stronger local planner so baseline behavior is still competitive and
+reproducible when the model is weak, unavailable, or partially aligned.
+"""
+
+from __future__ import annotations
+
+import ast
+from collections import Counter, defaultdict
+import hashlib
+import json
+import os
+import re
+import textwrap
+from typing import Any, Dict, Iterable, List, Mapping, Optional, Sequence, Tuple
+
+from openai import OpenAI
+
+from env import DataOpsEnv
+
+API_BASE_URL = os.getenv("API_BASE_URL", "https://router.huggingface.co/v1")
+MODEL_NAME = os.getenv("MODEL_NAME", "Qwen/Qwen3-VL-30B-A3B-Instruct:novita")
+HF_TOKEN = os.getenv("HF_TOKEN")
+BENCHMARK = os.getenv("OPENENV_BENCHMARK", "dataops-env")
+MAX_STEPS = 10
+TEMPERATURE = 0.0
+MAX_TOKENS = 160
+MODEL_RETRIES = 2
+FALLBACK_ACTION = "noop()"
+ACTION_PREFIX_RE = re.compile(r"^(action|next action)\s*[:\-]\s*", re.IGNORECASE)
+EMAIL_PATTERN = re.compile(r"^[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}$")
+TASK_ORDER = ["easy", "medium", "hard"]
+IDENTIFIER_COLUMNS = ("customer_id", "vendor_id", "partner_id")
+POLICY_CACHE_PATH = os.getenv("POLICY_CACHE_PATH", ".dataops_policy_cache.json")
+POLICY_CACHE_VERSION = 1
+
+SYSTEM_PROMPT = textwrap.dedent(
+    """
+    You control a data-cleaning environment.
+    Reply with exactly one action string and nothing else.
+
+    Only choose from the candidate actions provided by the user prompt.
+    Favor actions that remove visible issues quickly and avoid actions that were
+    already blocked because they caused errors or no progress.
+    Use single quotes for string arguments.
+    """
+).strip()
+
+
+class PolicyMemory:
+    """Persistent lightweight experience cache used across episodes and runs."""
+
+    def __init__(self, path: str) -> None:
+        self.path = path
+        self.data: Dict[str, Any] = {
+            "version": POLICY_CACHE_VERSION,
+            "states": {},
+            "patterns": {},
+        }
+        self._load()
+
+    def _load(self) -> None:
+        """Load cache from disk if it exists and is compatible."""
+
+        if not os.path.exists(self.path):
+            return
+        try:
+            with open(self.path, "r", encoding="utf-8") as handle:
+                payload = json.load(handle)
+        except (OSError, json.JSONDecodeError):
+            return
+        if not isinstance(payload, dict):
+            return
+        if int(payload.get("version", 0)) != POLICY_CACHE_VERSION:
+            return
+        self.data = payload
+
+    def save(self) -> None:
+        """Persist the current cache contents to disk."""
+
+        temp_path = f"{self.path}.tmp"
+        with open(temp_path, "w", encoding="utf-8") as handle:
+            json.dump(self.data, handle, indent=2, sort_keys=True)
+        os.replace(temp_path, self.path)
+
+    def _bucket(self, bucket_name: str, key: str) -> Dict[str, Any]:
+        """Return the mutable bucket for an exact state or a problem pattern."""
+
+        return self.data.setdefault(bucket_name, {}).setdefault(key, {"actions": {}})
+
+    def _action_stats(self, bucket_name: str, key: str, action_text: str) -> Dict[str, Any]:
+        """Return mutable stats for an action within a memory bucket."""
+
+        actions = self._bucket(bucket_name, key).setdefault("actions", {})
+        return actions.setdefault(
+            action_text,
+            {
+                "attempts": 0,
+                "successes": 0,
+                "progresses": 0,
+                "failures": 0,
+                "cumulative_reward": 0.0,
+                "last_error": None,
+            },
+        )
+
+    def update(
+        self,
+        *,
+        state_key: str,
+        pattern_key: str,
+        action_text: str,
+        reward: float,
+        progress_delta: float,
+        error: Optional[str],
+        done: bool,
+        task_score: float,
+    ) -> None:
+        """Record one step outcome for exact-state and problem-pattern memory."""
+
+        was_success = bool(done and task_score >= 0.95 and error is None)
+        made_progress = bool(progress_delta > 0.0 or reward > 0.0)
+        was_failure = bool(error is not None or (progress_delta == 0.0 and reward <= 0.0))
+
+        for bucket_name, key in (("states", state_key), ("patterns", pattern_key)):
+            stats = self._action_stats(bucket_name, key, action_text)
+            stats["attempts"] += 1
+            stats["cumulative_reward"] = round(
+                float(stats["cumulative_reward"]) + float(reward),
+                4,
+            )
+            stats["last_error"] = error
+            if was_success:
+                stats["successes"] += 1
+            elif made_progress:
+                stats["progresses"] += 1
+            if was_failure:
+                stats["failures"] += 1
+
+    def _combined_stats(self, state_key: str, pattern_key: str, action_text: str) -> Dict[str, float]:
+        """Merge exact-state and pattern-level stats into one weighted view."""
+
+        combined = {
+            "attempts": 0.0,
+            "successes": 0.0,
+            "progresses": 0.0,
+            "failures": 0.0,
+            "cumulative_reward": 0.0,
+        }
+        for bucket_name, key, weight in (
+            ("states", state_key, 1.0),
+            ("patterns", pattern_key, 0.5),
+        ):
+            stats = self.data.get(bucket_name, {}).get(key, {}).get("actions", {}).get(action_text)
+            if not isinstance(stats, dict):
+                continue
+            for field in combined:
+                combined[field] += float(stats.get(field, 0.0)) * weight
+        return combined
+
+    def score_action(self, state_key: str, pattern_key: str, action_text: str) -> float:
+        """Score a candidate action using remembered prior outcomes."""
+
+        stats = self._combined_stats(state_key, pattern_key, action_text)
+        attempts = max(1.0, stats["attempts"])
+        average_reward = stats["cumulative_reward"] / attempts
+        return round(
+            (stats["successes"] * 3.0)
+            + (stats["progresses"] * 1.25)
+            + average_reward
+            - (stats["failures"] * 2.0),
+            4,
+        )
+
+    def blocked_actions(self, state_key: str, pattern_key: str) -> set[str]:
+        """Return actions that repeatedly failed for the same state or pattern."""
+
+        blocked: set[str] = set()
+        for bucket_name, key in (("states", state_key), ("patterns", pattern_key)):
+            actions = self.data.get(bucket_name, {}).get(key, {}).get("actions", {})
+            for action_text, stats in actions.items():
+                attempts = int(stats.get("attempts", 0))
+                failures = int(stats.get("failures", 0))
+                successes = int(stats.get("successes", 0))
+                progresses = int(stats.get("progresses", 0))
+                if attempts >= 2 and failures >= attempts and successes == 0 and progresses == 0:
+                    blocked.add(action_text)
+        return blocked
+
+
+def log_start(task: str, env: str, model: str) -> None:
+    """Emit the required episode start line."""
+
+    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:
+    """Emit the required per-step line."""
+
+    error_value = error if error else "null"
+    print(
+        f"[STEP] step={step} action={action} reward={reward:.2f} "
+        f"done={str(done).lower()} error={error_value}",
+        flush=True,
+    )
+
+
+def log_end(success: bool, steps: int, rewards: List[float]) -> None:
+    """Emit the required episode end line."""
+
+    rewards_text = ",".join(f"{reward:.2f}" for reward in rewards)
+    print(
+        f"[END] success={str(success).lower()} steps={steps} rewards={rewards_text}",
+        flush=True,
+    )
+
+
+def build_history_lines(history: Sequence[str]) -> str:
+    """Render the last few steps for the model prompt."""
+
+    if not history:
+        return "None"
+    return "\n".join(history[-5:])
+
+
+def _stable_json(value: Any) -> str:
+    """Serialize a value deterministically for memory key generation."""
+
+    return json.dumps(value, sort_keys=True, separators=(",", ":"))
+
+
+def _hash_key(payload: Mapping[str, Any]) -> str:
+    """Build a compact deterministic memory key."""
+
+    return hashlib.sha1(_stable_json(payload).encode("utf-8")).hexdigest()
+
+
+def _normalize_issue_text(issue: str) -> str:
+    """Remove row-specific numbers so pattern memory generalizes better."""
+
+    lowered = issue.lower().strip()
+    return re.sub(r"\d+", "#", lowered)
+
+
+def _table_summary(table: Sequence[Mapping[str, Any]]) -> Dict[str, Any]:
+    """Extract a compact problem summary from the visible table state."""
+
+    present_columns = sorted({key for row in table for key in row.keys() if key != "row_id"})
+    missing_counts: Dict[str, int] = {}
+    for column in present_columns:
+        count = sum(1 for row in table if _is_missing(row.get(column)))
+        if count > 0:
+            missing_counts[column] = count
+
+    duplicate_counts: Dict[str, int] = {}
+    for column in list(IDENTIFIER_COLUMNS) + ["email"]:
+        values = [row.get(column) for row in table if row.get(column) not in (None, "")]
+        if values and len(values) != len(set(values)):
+            duplicate_counts[column] = len(values) - len(set(values))
+
+    return {
+        "row_count": len(table),
+        "present_columns": present_columns,
+        "missing_counts": missing_counts,
+        "duplicate_counts": duplicate_counts,
+        "invalid_email_count": sum(
+            1
+            for row in table
+            if row.get("email") not in (None, "") and not _is_valid_email(row.get("email"))
+        ),
+        "invalid_phone_count": sum(
+            1
+            for row in table
+            if row.get("phone") not in (None, "") and not _is_valid_phone(row.get("phone"))
+        ),
+        "title_case_columns": sorted(
+            column
+            for column in ("name", "city")
+            if any(_needs_title_case(row.get(column)) for row in table)
+        ),
+    }
+
+
+def build_memory_keys(
+    task_name: str,
+    task_variant: str,
+    goal: str,
+    observation: Mapping[str, Any],
+) -> Tuple[str, str]:
+    """Build exact-state and generalized problem-pattern keys."""
+
+    table = list(observation.get("table", []))
+    normalized_issues = sorted(_normalize_issue_text(str(issue)) for issue in observation.get("issues", []))
+    state_key = _hash_key(
+        {
+            "task_name": task_name,
+            "task_variant": task_variant,
+            "goal": goal,
+            "table": [
+                {key: row.get(key) for key in sorted(row.keys())}
+                for row in sorted(table, key=lambda row: int(row.get("row_id", 0)))
+            ],
+            "issues": normalized_issues,
+        }
+    )
+    pattern_key = _hash_key(
+        {
+            "task_name": task_name,
+            "goal": goal,
+            "summary": _table_summary(table),
+            "issues": normalized_issues,
+        }
+    )
+    return state_key, pattern_key
+
+
+def _is_missing(value: Any) -> bool:
+    """Return whether a value is missing."""
+
+    return value is None or value == ""
+
+
+def _needs_title_case(value: Any) -> bool:
+    """Return whether a string still needs title-case normalization."""
+
+    if not isinstance(value, str):
+        return False
+    cleaned = value.strip()
+    return bool(cleaned) and cleaned != cleaned.title()
+
+
+def _is_valid_email(value: Any) -> bool:
+    """Return whether an email-like string is valid."""
+
+    return isinstance(value, str) and bool(EMAIL_PATTERN.match(value.strip()))
+
+
+def _is_valid_phone(value: Any) -> bool:
+    """Return whether a phone-like string is valid."""
+
+    if not isinstance(value, str):
+        return False
+    digits = re.sub(r"\D", "", value)
+    return len(digits) == 10 or (len(digits) == 11 and digits.startswith("1"))
+
+
+def _slugify_text(value: str) -> str:
+    """Convert free text into a stable email-local-part fragment."""
+
+    lowered = re.sub(r"[^a-z0-9]+", ".", value.lower()).strip(".")
+    return lowered or "record"
+
+
+def _infer_email(row: Mapping[str, Any]) -> str:
+    """Infer a safe placeholder email from row context."""
+
+    if isinstance(row.get("name"), str) and row["name"].strip():
+        return f"{_slugify_text(row['name'])}@example.com"
+    for key in IDENTIFIER_COLUMNS:
+        if row.get(key):
+            return f"{str(row[key]).lower()}@example.com"
+    return f"row{row.get('row_id', 'unknown')}@example.com"
+
+
+def _infer_name(row: Mapping[str, Any]) -> str:
+    """Infer a readable name when a name field is missing."""
+
+    email = row.get("email")
+    if isinstance(email, str) and "@" in email:
+        return email.split("@", 1)[0].replace(".", " ").title()
+    for key in IDENTIFIER_COLUMNS:
+        if row.get(key):
+            return str(row[key]).replace("_", " ").title()
+    return "Unknown Record"
+
+
+def _infer_city(table: Sequence[Mapping[str, Any]]) -> str:
+    """Infer a plausible city using the mode of visible values."""
+
+    candidates = [
+        str(row.get("city")).title()
+        for row in table
+        if isinstance(row.get("city"), str) and row.get("city")
+    ]
+    if not candidates:
+        return "Seattle"
+    return Counter(candidates).most_common(1)[0][0]
+
+
+def _infer_fill_value(
+    row: Mapping[str, Any],
+    column: str,
+    table: Sequence[Mapping[str, Any]],
+) -> str:
+    """Infer a deterministic fill value from local table context."""
+
+    for key in IDENTIFIER_COLUMNS:
+        identifier = row.get(key)
+        if not identifier:
+            continue
+        for candidate in table:
+            if candidate.get("row_id") == row.get("row_id"):
+                continue
+            if candidate.get(key) == identifier and not _is_missing(candidate.get(column)):
+                return str(candidate[column])
+
+    if column == "email":
+        return _infer_email(row)
+    if column == "city":
+        return _infer_city(table)
+    if column == "phone":
+        return "555-555-0100"
+    if column == "status":
+        return "active"
+    if column == "name":
+        return _infer_name(row)
+    return "resolved"
+
+
+def _row_signature(row: Mapping[str, Any]) -> Tuple[Tuple[str, Any], ...]:
+    """Create a comparable row signature excluding runtime row identifiers."""
+
+    return tuple(sorted((key, value) for key, value in row.items() if key != "row_id"))
+
+
+def _build_action_string(payload: Mapping[str, Any]) -> str:
+    """Reconstruct a normalized action string for logging and filtering."""
+
+    action_type = str(payload["action_type"])
+    args: List[str] = []
+    for key in ("row_id", "column", "value"):
+        if key not in payload or payload[key] is None:
+            continue
+        value = payload[key]
+        if isinstance(value, str):
+            args.append(f"{key}='{value}'")
+        else:
+            args.append(f"{key}={value}")
+    return f"{action_type}({', '.join(args)})" if args else f"{action_type}()"
+
+
+def build_action_string(payload: Dict[str, Any]) -> str:
+    """Backward-compatible public wrapper around action string generation."""
+
+    return _build_action_string(payload)
+
+
+def parse_model_action(response_text: str) -> str:
+    """Extract a single action string from model output."""
+
+    if not response_text:
+        return FALLBACK_ACTION
+
+    for raw_line in response_text.splitlines():
+        line = ACTION_PREFIX_RE.sub("", raw_line.strip())
+        if "(" in line and line.endswith(")"):
+            return re.sub(r"\s+", " ", line)
+
+    compact = ACTION_PREFIX_RE.sub("", response_text.strip())
+    match = re.search(r"[a-zA-Z_]+\s*\(.*\)", compact)
+    if match:
+        return re.sub(r"\s+", " ", match.group(0))
+
+    return FALLBACK_ACTION
+
+
+def action_string_to_payload(action_str: str, step_number: int) -> Tuple[str, Dict[str, Any]]:
+    """Convert a model action string into an environment action payload."""
+
+    try:
+        expression = ast.parse(action_str, mode="eval").body
+    except SyntaxError:
+        return FALLBACK_ACTION, {"action_id": f"step-{step_number:03d}", "action_type": "noop"}
+
+    if not isinstance(expression, ast.Call) or not isinstance(expression.func, ast.Name):
+        return FALLBACK_ACTION, {"action_id": f"step-{step_number:03d}", "action_type": "noop"}
+
+    allowed_actions = {
+        "remove_duplicate",
+        "fill_missing",
+        "normalize_column",
+        "delete_row",
+        "validate",
+        "noop",
+    }
+    action_type = expression.func.id
+    if action_type not in allowed_actions:
+        return FALLBACK_ACTION, {"action_id": f"step-{step_number:03d}", "action_type": "noop"}
+
+    payload: Dict[str, Any] = {
+        "action_id": f"step-{step_number:03d}",
+        "action_type": action_type,
+    }
+    try:
+        for keyword in expression.keywords:
+            if keyword.arg is None:
+                continue
+            payload[keyword.arg] = ast.literal_eval(keyword.value)
+    except (SyntaxError, ValueError, TypeError):
+        return FALLBACK_ACTION, {"action_id": f"step-{step_number:03d}", "action_type": "noop"}
+
+    return _build_action_string(payload), payload
+
+
+def create_client() -> Optional[OpenAI]:
+    """Create an OpenAI-compatible client when credentials look real."""
+
+    if HF_TOKEN in {None, "", "local-test", "test", "dummy"}:
+        return None
+    return OpenAI(base_url=API_BASE_URL, api_key=HF_TOKEN)
+
+
+def _extract_response_text(content: Any) -> str:
+    """Normalize OpenAI response content into plain text."""
+
+    if isinstance(content, str):
+        return content
+    if isinstance(content, list):
+        return "".join(
+            str(part.get("text", ""))
+            for part in content
+            if isinstance(part, dict)
+        )
+    return str(content or "")
+
+
+def _table_preview(table: Sequence[Mapping[str, Any]], limit: int = 6) -> str:
+    """Render a compact table preview for prompting."""
+
+    preview_lines: List[str] = []
+    for row in table[:limit]:
+        summary = ", ".join(
+            f"{key}={value}"
+            for key, value in row.items()
+            if key in {"row_id", "name", "city", "email", "phone", "status", "customer_id", "vendor_id", "partner_id"}
+        )
+        preview_lines.append(f"- {summary}")
+    return "\n".join(preview_lines) if preview_lines else "- None"
+
+
+def build_user_prompt(
+    step: int,
+    goal: str,
+    observation: Dict[str, Any],
+    history: Sequence[str],
+    last_error: Optional[str],
+    candidate_actions: Sequence[str],
+    blocked_actions: Sequence[str],
+) -> str:
+    """Construct a compact prompt that constrains the model to useful actions."""
+
+    issues = observation.get("issues", [])
+    hints = observation.get("hints", [])
+    issues_text = "\n".join(f"- {issue}" for issue in issues[:6]) if issues else "- None"
+    hints_text = "\n".join(f"- {hint}" for hint in hints[:3]) if hints else "- None"
+    candidates_text = "\n".join(f"- {action}" for action in candidate_actions)
+    blocked_text = "\n".join(f"- {action}" for action in blocked_actions[:5]) if blocked_actions else "- None"
+
+    return textwrap.dedent(
+        f"""
+        Step: {step}
+        Goal: {goal}
+        Steps remaining: {observation.get("steps_remaining")}
+        Progress: {observation.get("progress")}
+        Current issues:
+        {issues_text}
+        Current hints:
+        {hints_text}
+        Table preview:
+        {_table_preview(observation.get("table", []))}
+        Recent history:
+        {build_history_lines(history)}
+        Last action error: {last_error or "null"}
+        Blocked actions:
+        {blocked_text}
+
+        Choose exactly one action from this candidate list:
+        {candidates_text}
+        """
+    ).strip()
+
+
+def _prefer_action(
+    candidates: Sequence[Dict[str, Any]],
+    blocked_actions: set[str],
+) -> Dict[str, Any]:
+    """Return the first candidate action that is not blocked."""
+
+    for candidate in candidates:
+        action_text = _build_action_string(candidate)
+        if action_text not in blocked_actions:
+            return dict(candidate)
+    return {"action_type": "validate"}
+
+
+def _exact_duplicate_candidates(table: Sequence[Mapping[str, Any]]) -> List[Dict[str, Any]]:
+    """Generate explicit remove-duplicate actions for exact duplicate rows."""
+
+    groups: Dict[Tuple[Tuple[str, Any], ...], List[int]] = defaultdict(list)
+    for row in table:
+        row_id = row.get("row_id")
+        if row_id is None:
+            continue
+        groups[_row_signature(row)].append(int(row_id))
+
+    actions: List[Dict[str, Any]] = []
+    for row_ids in groups.values():
+        if len(row_ids) > 1:
+            actions.append({"action_type": "remove_duplicate", "row_id": max(row_ids)})
+    return actions
+
+
+def _group_by_identifier(table: Sequence[Mapping[str, Any]]) -> Dict[Tuple[str, str], List[Dict[str, Any]]]:
+    """Group rows by likely business identifiers."""
+
+    groups: Dict[Tuple[str, str], List[Dict[str, Any]]] = defaultdict(list)
+    for row in table:
+        for key in IDENTIFIER_COLUMNS:
+            value = row.get(key)
+            if value not in (None, ""):
+                groups[(key, str(value))].append(dict(row))
+    return groups
+
+
+def _row_quality_score(row: Mapping[str, Any]) -> int:
+    """Score a row so lower-quality conflict rows can be removed first."""
+
+    score = 0
+    if _is_valid_email(row.get("email")):
+        score += 3
+    if _is_valid_phone(row.get("phone")) or row.get("phone") in (None, ""):
+        score += 2
+    if isinstance(row.get("status"), str) and row.get("status") == "active":
+        score += 1
+    if isinstance(row.get("name"), str) and row.get("name").strip():
+        score += 1
+    return score
+
+
+def _structural_delete_candidates(table: Sequence[Mapping[str, Any]]) -> List[Dict[str, Any]]:
+    """Generate delete actions for non-exact structural conflicts."""
+
+    actions: List[Dict[str, Any]] = []
+    for rows in _group_by_identifier(table).values():
+        if len(rows) < 2:
+            continue
+        signatures = {_row_signature(row) for row in rows}
+        if len(signatures) == 1:
+            continue
+        worst_row = sorted(
+            rows,
+            key=lambda row: (_row_quality_score(row), int(row.get("row_id", 0))),
+        )[0]
+        actions.append({"action_type": "delete_row", "row_id": int(worst_row["row_id"])})
+
+    email_groups: Dict[str, List[Dict[str, Any]]] = defaultdict(list)
+    for row in table:
+        email = row.get("email")
+        if email not in (None, ""):
+            email_groups[str(email)].append(dict(row))
+    for rows in email_groups.values():
+        if len(rows) < 2:
+            continue
+        worst_row = sorted(
+            rows,
+            key=lambda row: (_row_quality_score(row), int(row.get("row_id", 0))),
+        )[0]
+        action = {"action_type": "delete_row", "row_id": int(worst_row["row_id"])}
+        if action not in actions:
+            actions.append(action)
+    return actions
+
+
+def _missing_value_candidates(table: Sequence[Mapping[str, Any]]) -> List[Dict[str, Any]]:
+    """Generate candidate fill actions for visible missing values."""
+
+    present_columns = {key for row in table for key in row.keys()}
+    priorities = [
+        column
+        for column in ["email", "city", "phone", "status", "name"]
+        if column in present_columns
+    ]
+    actions: List[Dict[str, Any]] = []
+    for column in priorities:
+        for row in table:
+            if _is_missing(row.get(column)):
+                actions.append(
+                    {
+                        "action_type": "fill_missing",
+                        "row_id": int(row["row_id"]),
+                        "column": column,
+                        "value": _infer_fill_value(row, column, table),
+                    }
+                )
+    return actions
+
+
+def _normalization_candidates(table: Sequence[Mapping[str, Any]]) -> List[Dict[str, Any]]:
+    """Generate candidate column-normalization actions."""
+
+    candidates: List[Dict[str, Any]] = []
+    if any(row.get("email") not in (None, "") and not _is_valid_email(row.get("email")) for row in table):
+        candidates.append({"action_type": "normalize_column", "column": "email"})
+    if any(row.get("phone") not in (None, "") and not _is_valid_phone(row.get("phone")) for row in table):
+        candidates.append({"action_type": "normalize_column", "column": "phone"})
+    if any(_needs_title_case(row.get("name")) for row in table):
+        candidates.append({"action_type": "normalize_column", "column": "name"})
+    if any(_needs_title_case(row.get("city")) for row in table):
+        candidates.append({"action_type": "normalize_column", "column": "city"})
+    return candidates
+
+
+def propose_candidate_actions(
+    observation: Mapping[str, Any],
+    blocked_actions: set[str],
+) -> List[Dict[str, Any]]:
+    """Generate ranked candidate actions from visible table state."""
+
+    table = list(observation.get("table", []))
+    candidates = (
+        _exact_duplicate_candidates(table)
+        + _structural_delete_candidates(table)
+        + _missing_value_candidates(table)
+        + _normalization_candidates(table)
+        + [{"action_type": "validate"}]
+        + [{"action_type": "noop"}]
+    )
+
+    unique_candidates: List[Dict[str, Any]] = []
+    seen: set[str] = set()
+    for candidate in candidates:
+        action_text = _build_action_string(candidate)
+        if action_text in seen:
+            continue
+        seen.add(action_text)
+        unique_candidates.append(candidate)
+
+    preferred = _prefer_action(unique_candidates, blocked_actions)
+    preferred_text = _build_action_string(preferred)
+    ordered = [preferred] + [
+        candidate
+        for candidate in unique_candidates
+        if _build_action_string(candidate) != preferred_text
+    ]
+    return ordered[:8]
+
+
+def _order_candidates_with_memory(
+    candidates: Sequence[Dict[str, Any]],
+    memory: PolicyMemory,
+    state_key: str,
+    pattern_key: str,
+) -> List[Dict[str, Any]]:
+    """Re-rank candidates using persistent cross-episode memory."""
+
+    scored = []
+    for index, candidate in enumerate(candidates):
+        action_text = _build_action_string(candidate)
+        scored.append(
+            (
+                -memory.score_action(state_key, pattern_key, action_text),
+                index,
+                dict(candidate),
+            )
+        )
+    scored.sort(key=lambda item: (item[0], item[1]))
+    return [item[2] for item in scored]
+
+
+def model_action(
+    client: Optional[OpenAI],
+    model_name: str,
+    step: int,
+    goal: str,
+    observation: Dict[str, Any],
+    history: Sequence[str],
+    last_error: Optional[str],
+    candidate_actions: Sequence[str],
+    blocked_actions: Sequence[str],
+) -> Optional[str]:
+    """Ask the model to choose among pre-computed candidate actions."""
+
+    if client is None:
+        return None
+
+    prompt = build_user_prompt(
+        step=step,
+        goal=goal,
+        observation=observation,
+        history=history,
+        last_error=last_error,
+        candidate_actions=candidate_actions,
+        blocked_actions=blocked_actions,
+    )
+    current_prompt = prompt
+    candidate_set = set(candidate_actions)
+    for _ in range(MODEL_RETRIES):
+        try:
+            completion = client.chat.completions.create(
+                model=model_name,
+                messages=[
+                    {
+                        "role": "system",
+                        "content": [{"type": "text", "text": SYSTEM_PROMPT}],
+                    },
+                    {
+                        "role": "user",
+                        "content": [{"type": "text", "text": current_prompt}],
+                    },
+                ],
+                temperature=TEMPERATURE,
+                max_tokens=MAX_TOKENS,
+                stream=False,
+            )
+            response_text = _extract_response_text(completion.choices[0].message.content)
+            action_text = parse_model_action(response_text)
+            if action_text in candidate_set and action_text not in set(blocked_actions):
+                return action_text
+            current_prompt = (
+                prompt
+                + "\n\nYour previous answer was invalid or blocked. Choose exactly one action from the candidate list."
+            )
+        except Exception:  # noqa: BLE001
+            return None
+    return None
+
+
+def choose_action(
+    client: Optional[OpenAI],
+    memory: PolicyMemory,
+    task_name: str,
+    task_variant: str,
+    observation: Dict[str, Any],
+    goal: str,
+    step_number: int,
+    history: Sequence[str],
+    last_error: Optional[str],
+    blocked_actions: set[str],
+) -> Tuple[str, Dict[str, Any], str, str, str]:
+    """Choose the next action using a heuristic planner with optional model arbitration."""
+
+    state_key, pattern_key = build_memory_keys(task_name, task_variant, goal, observation)
+    memory_blocked = memory.blocked_actions(state_key, pattern_key)
+    combined_blocked = set(blocked_actions) | set(memory_blocked)
+    candidates = propose_candidate_actions(observation, combined_blocked)
+    candidates = _order_candidates_with_memory(candidates, memory, state_key, pattern_key)
+    heuristic_candidate = candidates[0]
+    heuristic_text = _build_action_string(heuristic_candidate)
+    candidate_texts = [_build_action_string(candidate) for candidate in candidates]
+
+    model_text = model_action(
+        client=client,
+        model_name=MODEL_NAME,
+        step=step_number,
+        goal=goal,
+        observation=observation,
+        history=history,
+        last_error=last_error,
+        candidate_actions=candidate_texts,
+        blocked_actions=sorted(combined_blocked),
+    )
+
+    chosen_text = model_text or heuristic_text
+    normalized_text, payload = action_string_to_payload(chosen_text, step_number)
+    if normalized_text in combined_blocked:
+        normalized_text, payload = action_string_to_payload(heuristic_text, step_number)
+        return normalized_text, payload, "heuristic", state_key, pattern_key
+    return normalized_text, payload, "model" if model_text else "heuristic", state_key, pattern_key
+
+
+def run_episode(
+    client: Optional[OpenAI],
+    memory: PolicyMemory,
+    task_name: str,
+    seed: int,
+) -> float:
+    """Run one deterministic task episode and return its final task score."""
+
+    env = DataOpsEnv(seed=seed, task_name=task_name)
+    rewards: List[float] = []
+    history: List[str] = []
+    blocked_actions: set[str] = set()
+    steps_taken = 0
+    success = False
+    last_error: Optional[str] = None
+    final_score = 0.0
+    task_variant = "unknown"
+
+    try:
+        log_start(task=task_name, env=BENCHMARK, model=MODEL_NAME)
+        observation_model = env.reset()
+        observation = observation_model.model_dump()
+        task_variant = str(env.state().get("task_variant", "unknown"))
+
+        for step_number in range(1, MAX_STEPS + 1):
+            action_text, action_payload, action_source, state_key, pattern_key = choose_action(
+                client=client,
+                memory=memory,
+                task_name=task_name,
+                task_variant=task_variant,
+                observation=observation,
+                goal=observation_model.goal,
+                step_number=step_number,
+                history=history,
+                last_error=last_error,
+                blocked_actions=blocked_actions,
+            )
+
+            try:
+                observation_model, reward, done, info = env.step(action_payload)
+                observation = observation_model.model_dump()
+                result = info.get("result", {})
+                progress_delta = float(result.get("progress_delta", 0.0))
+                error_value = result.get("error_type") or info.get("error") or None
+                final_score = float(info.get("task_score", 0.0))
+                if error_value == "general":
+                    error_value = None
+                memory.update(
+                    state_key=state_key,
+                    pattern_key=pattern_key,
+                    action_text=action_text,
+                    reward=reward,
+                    progress_delta=progress_delta,
+                    error=error_value,
+                    done=done,
+                    task_score=final_score,
+                )
+                if error_value or progress_delta == 0.0 or reward <= 0.0:
+                    blocked_actions.add(action_text)
+            except Exception as exc:  # noqa: BLE001
+                reward = 0.0
+                done = True
+                info = {}
+                error_value = str(exc)
+                blocked_actions.add(action_text)
+                memory.update(
+                    state_key=state_key,
+                    pattern_key=pattern_key,
+                    action_text=action_text,
+                    reward=reward,
+                    progress_delta=0.0,
+                    error=error_value,
+                    done=done,
+                    task_score=final_score,
+                )
+
+            rewards.append(reward)
+            steps_taken = step_number
+            last_error = error_value
+            log_step(
+                step=step_number,
+                action=action_text,
+                reward=reward,
+                done=done,
+                error=error_value,
+            )
+
+            history.append(
+                f"step={step_number} source={action_source} action={action_text} "
+                f"reward={reward:.2f} done={str(done).lower()} error={error_value or 'null'}"
+            )
+
+            if done:
+                success = bool(final_score >= 0.95 and error_value is None)
+                break
+    finally:
+        memory.save()
+        close_method = getattr(env, "close", None)
+        if callable(close_method):
+            close_method()
+        log_end(success=success, steps=steps_taken, rewards=rewards)
+    return final_score
+
+
+def main() -> None:
+    """Run all benchmark tasks with deterministic ordering and stdout formatting."""
+
+    client = create_client()
+    memory = PolicyMemory(POLICY_CACHE_PATH)
+    for task_index, task_name in enumerate(TASK_ORDER):
+        run_episode(client=client, memory=memory, task_name=task_name, seed=task_index)
+
+
+if __name__ == "__main__":
+    main()

models.py

@@ -0,0 +1,130 @@
+"""Shared data models for ``dataops-gym``.
+
+This module is responsible for defining typed request, response, and domain
+schemas used across task execution, inference, grading, and server layers.
+"""
+
+from __future__ import annotations
+
+from typing import Any, Dict, List, Literal, Optional
+
+from pydantic import BaseModel, Field, root_validator
+
+
+class Action(BaseModel):
+    """Represents a single environment action issued by an agent or client."""
+
+    class Config:
+        arbitrary_types_allowed = True
+        extra = "forbid"
+
+    action_id: Optional[str] = Field(
+        default=None,
+        description="A unique identifier for the action instance, useful for tracking repeated actions and mistake patterns.",
+    )
+    action_type: Literal[
+        "remove_duplicate",
+        "fill_missing",
+        "normalize_column",
+        "delete_row",
+        "validate",
+        "noop",
+    ] = Field(
+        ...,
+        description="The type of data-cleaning action to apply in the environment.",
+    )
+    column: Optional[str] = Field(
+        default=None,
+        description="Optional target column name associated with the action.",
+    )
+    row_id: Optional[int] = Field(
+        default=None,
+        description="Optional target row identifier associated with the action.",
+    )
+    value: Optional[str] = Field(
+        default=None,
+        description="Optional value payload used by the action when needed.",
+    )
+
+    @root_validator(skip_on_failure=True)
+    def validate_action_requirements(cls, values: Dict[str, Any]) -> Dict[str, Any]:
+        """Enforce conditional field requirements for specific action types."""
+        action_type = values.get("action_type")
+        column = values.get("column")
+        row_id = values.get("row_id")
+        value = values.get("value")
+
+        if action_type == "delete_row" and row_id is None:
+            raise ValueError("row_id must not be None when action_type is 'delete_row'")
+        if action_type == "normalize_column" and column is None:
+            raise ValueError("column must not be None when action_type is 'normalize_column'")
+        if action_type == "fill_missing" and (column is None or value is None):
+            raise ValueError(
+                "column and value must not be None when action_type is 'fill_missing'"
+            )
+
+        return values
+
+
+class Observation(BaseModel):
+    """Represents the current observable state returned by the environment."""
+
+    class Config:
+        arbitrary_types_allowed = True
+        extra = "forbid"
+
+    goal: str = Field(
+        ...,
+        description="A natural language description of the task objective the agent must achieve.",
+    )
+    table: List[Dict[str, Any]] = Field(
+        ...,
+        description="JSON-serializable table snapshot represented as a list of row dictionaries.",
+    )
+    issues: List[str] = Field(
+        ...,
+        description="Detected data-quality issues currently present in the table.",
+    )
+    history: List[str] = Field(
+        ...,
+        description="Ordered list of previously applied actions or events.",
+    )
+    mistakes: Dict[str, int] = Field(
+        ...,
+        description="Counts of mistake categories accumulated during the episode.",
+    )
+    hints: List[str] = Field(
+        ...,
+        description="Optional guidance hints available to the agent or client.",
+    )
+    progress: float = Field(
+        ...,
+        ge=0.0,
+        le=1.0,
+        description="A normalized estimate (0.0–1.0) of how much of the task is completed.",
+    )
+    steps_remaining: int = Field(
+        ...,
+        description="Number of steps remaining before the episode terminates.",
+    )
+
+
+class Reward(BaseModel):
+    """Represents the reward outcome associated with an environment step."""
+
+    class Config:
+        arbitrary_types_allowed = True
+        extra = "forbid"
+
+    reward: float = Field(
+        ...,
+        description="Numeric reward assigned to the most recent action or transition.",
+    )
+    reason: str = Field(
+        ...,
+        description="Human-readable explanation for why the reward was assigned.",
+    )
+    components: Dict[str, float] = Field(
+        ...,
+        description="Breakdown of reward contributions (e.g., duplicate_removal: 0.3, penalty: -0.1)",
+    )

openenv.yaml

@@ -0,0 +1,16 @@
+# OpenEnv configuration for `dataops-env`.
+# Responsibility: declare minimal environment metadata and task registration.
+
+name: dataops-env
+description: Multi-step enterprise data cleaning environment
+version: "1.0.0"
+runtime:
+  python: "3.10+"
+entrypoint: env:DataOpsEnv
+tasks:
+  easy:
+    factory: task:easy_cleaning_task
+  medium:
+    factory: task:medium_normalization_task
+  hard:
+    factory: task:hard_conflict_resolution_task

pyproject.toml

@@ -0,0 +1,24 @@
+[build-system]
+requires = ["setuptools>=68", "wheel"]
+build-backend = "setuptools.build_meta"
+
+[project]
+name = "dataops-env"
+version = "1.0.0"
+description = "Multi-step enterprise data cleaning OpenEnv environment."
+readme = "README.md"
+requires-python = ">=3.10"
+dependencies = [
+    "fastapi",
+    "numpy",
+    "openenv-core>=0.2.0",
+    "openai",
+    "pydantic",
+    "uvicorn",
+]
+
+[project.scripts]
+server = "server.app:main"
+
+[tool.setuptools]
+py-modules = ["env", "grader", "inference", "models", "task"]

requirements.txt

@@ -0,0 +1,6 @@
+fastapi
+numpy
+openenv-core>=0.2.0
+openai
+pydantic
+uvicorn

server/__init__.py

@@ -0,0 +1 @@
+"""Server package for the dataops OpenEnv environment."""

server/app.py

@@ -0,0 +1,156 @@
+"""Application server bootstrap for ``dataops-gym``.
+
+This module is responsible for exposing runtime APIs, health endpoints, and
+deployment-facing application setup for the environment.
+"""
+
+from __future__ import annotations
+
+import logging
+import os
+from pathlib import Path
+import sys
+from threading import RLock
+from typing import Any, Dict, Optional
+
+from fastapi import FastAPI, HTTPException, Request
+from fastapi.responses import JSONResponse
+from pydantic import BaseModel, Field
+import uvicorn
+
+
+PROJECT_ROOT = Path(__file__).resolve().parent.parent
+if str(PROJECT_ROOT) not in sys.path:
+    sys.path.insert(0, str(PROJECT_ROOT))
+
+from env import DataOpsEnv
+from models import Action
+
+
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+app = FastAPI(title="dataops-env", version="1.0.0")
+active_env: Optional[DataOpsEnv] = None
+active_env_lock = RLock()
+
+
+class ResetRequest(BaseModel):
+    """Optional reset controls for reproducible task selection."""
+
+    seed: int = Field(default=0, description="Deterministic seed for task sampling.")
+    task_name: str | None = Field(
+        default=None,
+        description="Optional fixed task name: easy, medium, or hard.",
+    )
+
+
+@app.exception_handler(Exception)
+async def unhandled_exception_handler(
+    request: Request, exc: Exception
+) -> JSONResponse:
+    """Return a safe error payload for unexpected server failures."""
+
+    logger.exception("Unhandled server error on %s", request.url.path, exc_info=exc)
+    return JSONResponse(
+        status_code=500,
+        content={"detail": "Internal server error"},
+    )
+
+
+@app.get("/health")
+def health() -> Dict[str, str]:
+    """Return a lightweight deployment health signal."""
+
+    return {"status": "healthy"}
+
+
+@app.post("/reset")
+def reset(payload: ResetRequest | None = None) -> Dict[str, Any]:
+    """Reset the environment and return the initial observation."""
+
+    try:
+        request = payload or ResetRequest()
+        env = DataOpsEnv(seed=request.seed, task_name=request.task_name)
+        observation = env.reset()
+
+        global active_env
+        with active_env_lock:
+            previous_env = active_env
+            active_env = env
+
+        return {
+            "task_name": env.state().get("task_name"),
+            "observation": observation.model_dump(),
+        }
+    except Exception as exc:
+        logger.exception("Failed to reset environment", exc_info=exc)
+        raise HTTPException(status_code=500, detail="Failed to reset environment") from exc
+    finally:
+        if "previous_env" in locals() and previous_env is not None:
+            close_method = getattr(previous_env, "close", None)
+            if callable(close_method):
+                close_method()
+
+
+@app.post("/step")
+def step(action: Action) -> Dict[str, Any]:
+    """Apply a single action to the environment and return the step result."""
+
+    try:
+        with active_env_lock:
+            if active_env is None:
+                raise HTTPException(
+                    status_code=400,
+                    detail="Environment not initialized. Call /reset first.",
+                )
+            observation, reward, done, info = active_env.step(action)
+
+        return {
+            "observation": observation.model_dump(),
+            "reward": reward,
+            "done": done,
+            "info": info,
+        }
+    except HTTPException:
+        raise
+    except RuntimeError as exc:
+        raise HTTPException(status_code=400, detail=str(exc)) from exc
+    except ValueError as exc:
+        raise HTTPException(status_code=422, detail=str(exc)) from exc
+    except Exception as exc:
+        logger.exception("Failed to execute environment step", exc_info=exc)
+        raise HTTPException(status_code=500, detail="Failed to execute step") from exc
+
+
+@app.get("/state")
+def state() -> Dict[str, Any]:
+    """Return the current internal environment state as JSON."""
+
+    try:
+        with active_env_lock:
+            if active_env is None:
+                raise HTTPException(
+                    status_code=400,
+                    detail="Environment not initialized. Call /reset first.",
+                )
+            return active_env.state()
+    except HTTPException:
+        raise
+    except Exception as exc:
+        logger.exception("Failed to fetch environment state", exc_info=exc)
+        raise HTTPException(status_code=500, detail="Failed to fetch state") from exc
+
+
+def main() -> None:
+    """Run the FastAPI application with uvicorn."""
+
+    uvicorn.run(
+        "server.app:app",
+        host="0.0.0.0",
+        port=int(os.getenv("PORT", "7860")),
+    )
+
+
+if __name__ == "__main__":
+    main()

task.py

@@ -0,0 +1,463 @@
+"""Task definitions for ``dataops-gym``.
+
+This module defines the benchmark scenarios used by the OpenEnv environment.
+Each public task family keeps the hackathon-facing `easy` / `medium` / `hard`
+shape while internally supporting deterministic variants so the benchmark is
+broader and less gameable.
+"""
+
+from __future__ import annotations
+
+from typing import Any, Dict, List, TypedDict
+
+
+TableRow = Dict[str, Any]
+
+
+class TaskDefinition(TypedDict):
+    """Typed structure returned by task factory functions."""
+
+    initial_table: List[TableRow]
+    hidden_issues: List[Dict[str, Any]]
+    constraints: List[str]
+    max_steps: int
+    goal: str
+    difficulty: str
+    required_columns: List[str]
+    expected_outcome: Dict[str, Any]
+    variant_id: str
+
+
+class HiddenIssue(TypedDict, total=False):
+    """Structured hidden issue description for a task table."""
+
+    type: str
+    rows: List[int]
+    row: int
+    column: str
+    constraint: str
+    description: str
+
+
+def _pick_variant(variant: int | None, variants: List[TaskDefinition]) -> TaskDefinition:
+    """Select a deterministic task variant with a stable default."""
+
+    index = 0 if variant is None else max(0, min(len(variants) - 1, int(variant)))
+    return variants[index]
+
+
+def easy_cleaning_task(variant: int | None = None) -> TaskDefinition:
+    """Create an easy multi-step cleaning task with duplicates and missing data."""
+
+    variants: List[TaskDefinition] = [
+        {
+            "goal": "Clean the dataset by removing duplicates and filling missing values.",
+            "difficulty": "easy",
+            "variant_id": "easy_customer_master",
+            "required_columns": ["name", "city", "email"],
+            "expected_outcome": {
+                "expected_row_count": 4,
+                "required_non_null_columns": ["name", "city", "email"],
+                "unique_by": ["customer_id"],
+                "exactly_one_of_rows": [[2, 3]],
+                "validation_rules": [
+                    "Exactly one of rows 2 or 3 should remain after deduplication.",
+                    "No remaining row should have null values in name, city, or email.",
+                    "All remaining customer_id values should be unique.",
+                ],
+            },
+            "initial_table": [
+                {"row_id": 1, "customer_id": "C001", "name": "Alice Wong", "city": "Seattle", "email": "alice@example.com"},
+                {"row_id": 2, "customer_id": "C002", "name": "Ben Ortiz", "city": None, "email": "ben@example.com"},
+                {"row_id": 3, "customer_id": "C002", "name": "Ben Ortiz", "city": None, "email": "ben@example.com"},
+                {"row_id": 4, "customer_id": "C003", "name": "Carla Singh", "city": "Austin", "email": None},
+                {"row_id": 5, "customer_id": "C004", "name": "Drew Park", "city": "Boston", "email": "drew@example.com"},
+            ],
+            "hidden_issues": [
+                {
+                    "type": "duplicate",
+                    "rows": [2, 3],
+                    "description": "Rows 2 and 3 are duplicates and only one should remain.",
+                },
+                {
+                    "type": "missing_value",
+                    "row": 2,
+                    "column": "city",
+                    "description": "Row 2 is missing a required city value.",
+                },
+                {
+                    "type": "missing_value",
+                    "row": 4,
+                    "column": "email",
+                    "description": "Row 4 is missing a required email value.",
+                },
+            ],
+            "constraints": [
+                "Keep one representative row for each real customer.",
+                "Do not delete rows solely because they contain missing values.",
+                "Name, city, and email must be populated for every remaining row.",
+            ],
+            "max_steps": 7,
+        },
+        {
+            "goal": "Clean the dataset by removing duplicates and filling missing values.",
+            "difficulty": "easy",
+            "variant_id": "easy_vendor_onboarding",
+            "required_columns": ["name", "city", "email"],
+            "expected_outcome": {
+                "expected_row_count": 4,
+                "required_non_null_columns": ["name", "city", "email"],
+                "unique_by": ["vendor_id"],
+                "exactly_one_of_rows": [[32, 33]],
+                "validation_rules": [
+                    "Exactly one of rows 32 or 33 should remain after deduplication.",
+                    "No remaining row should have null values in name, city, or email.",
+                    "All remaining vendor_id values should be unique.",
+                ],
+            },
+            "initial_table": [
+                {"row_id": 31, "vendor_id": "V001", "name": "Northwind Foods", "city": "Denver", "email": "ops@northwind.example.com"},
+                {"row_id": 32, "vendor_id": "V002", "name": "Blue Harbor Ltd", "city": "Miami", "email": "contact@blueharbor.example.com"},
+                {"row_id": 33, "vendor_id": "V002", "name": "Blue Harbor Ltd", "city": "Miami", "email": "contact@blueharbor.example.com"},
+                {"row_id": 34, "vendor_id": "V003", "name": "Atlas Office Supply", "city": None, "email": "service@atlas.example.com"},
+                {"row_id": 35, "vendor_id": "V004", "name": "Peak Systems", "city": "Portland", "email": None},
+            ],
+            "hidden_issues": [
+                {
+                    "type": "duplicate",
+                    "rows": [32, 33],
+                    "description": "Rows 32 and 33 are duplicates and only one should remain.",
+                },
+                {
+                    "type": "missing_value",
+                    "row": 34,
+                    "column": "city",
+                    "description": "Row 34 is missing a required city value.",
+                },
+                {
+                    "type": "missing_value",
+                    "row": 35,
+                    "column": "email",
+                    "description": "Row 35 is missing a required email value.",
+                },
+            ],
+            "constraints": [
+                "Keep one representative row for each real vendor.",
+                "Do not delete rows solely because they contain missing values.",
+                "Name, city, and email must be populated for every remaining row.",
+            ],
+            "max_steps": 7,
+        },
+    ]
+    return _pick_variant(variant, variants)
+
+
+def medium_normalization_task(variant: int | None = None) -> TaskDefinition:
+    """Create a medium multi-step normalization task with several issue types."""
+
+    variants: List[TaskDefinition] = [
+        {
+            "goal": "Normalize the dataset by fixing casing, removing duplicates, and correcting invalid email formats.",
+            "difficulty": "medium",
+            "variant_id": "medium_customer_normalization",
+            "required_columns": ["name", "city", "email"],
+            "expected_outcome": {
+                "expected_row_count": 5,
+                "required_non_null_columns": ["name", "city", "email"],
+                "unique_by": ["customer_id"],
+                "normalized_columns": {"name": "title_case", "city": "title_case"},
+                "format_rules": {"email": "valid_email"},
+                "exactly_one_of_rows": [[11, 13]],
+                "validation_rules": [
+                    "Exactly one of rows 11 or 13 should remain after deduplication.",
+                    "All remaining emails should satisfy a valid email format.",
+                    "Names and cities should follow a consistent human-readable casing convention.",
+                    "All remaining customer_id values should be unique.",
+                ],
+            },
+            "initial_table": [
+                {"row_id": 10, "customer_id": "C100", "name": "jane miller", "city": "new york", "email": "jane.miller@example.com"},
+                {"row_id": 11, "customer_id": "C101", "name": "OMAR HASSAN", "city": "CHICAGO", "email": "omar.hassan[at]example.com"},
+                {"row_id": 12, "customer_id": "C102", "name": "Priya Nair", "city": "San Jose", "email": "priya.nair@example.com"},
+                {"row_id": 13, "customer_id": "C101", "name": "OMAR HASSAN", "city": "CHICAGO", "email": "omar.hassan[at]example.com"},
+                {"row_id": 14, "customer_id": "C103", "name": "li wei", "city": "seattle", "email": "li.wei.example.com"},
+                {"row_id": 15, "customer_id": "C104", "name": "Maria Gomez", "city": "Austin", "email": "maria.gomez@example.com"},
+            ],
+            "hidden_issues": [
+                {
+                    "type": "duplicate",
+                    "rows": [11, 13],
+                    "description": "Rows 11 and 13 are duplicates and only one should remain.",
+                },
+                {
+                    "type": "inconsistent_casing",
+                    "rows": [10, 11, 14],
+                    "column": "name",
+                    "description": "Rows 10, 11, and 14 contain inconsistent casing in names.",
+                },
+                {
+                    "type": "inconsistent_casing",
+                    "rows": [10, 11, 14],
+                    "column": "city",
+                    "description": "Rows 10, 11, and 14 contain inconsistent casing in cities.",
+                },
+                {
+                    "type": "invalid_format",
+                    "row": 11,
+                    "column": "email",
+                    "description": "Row 11 contains an invalid email format.",
+                },
+                {
+                    "type": "invalid_format",
+                    "row": 14,
+                    "column": "email",
+                    "description": "Row 14 contains an invalid email format.",
+                },
+            ],
+            "constraints": [
+                "Preserve the original entity identity of each remaining row.",
+                "Normalize names and cities to a consistent human-readable casing style.",
+                "Only repair emails that are actually invalid.",
+                "Do not introduce new duplicates while normalizing values.",
+            ],
+            "max_steps": 9,
+        },
+        {
+            "goal": "Normalize the dataset by fixing casing, removing duplicates, and correcting invalid email formats.",
+            "difficulty": "medium",
+            "variant_id": "medium_partner_directory",
+            "required_columns": ["name", "city", "email"],
+            "expected_outcome": {
+                "expected_row_count": 5,
+                "required_non_null_columns": ["name", "city", "email"],
+                "unique_by": ["partner_id"],
+                "normalized_columns": {"name": "title_case", "city": "title_case"},
+                "format_rules": {"email": "valid_email"},
+                "exactly_one_of_rows": [[41, 43]],
+                "validation_rules": [
+                    "Exactly one of rows 41 or 43 should remain after deduplication.",
+                    "All remaining emails should satisfy a valid email format.",
+                    "Names and cities should use consistent title case.",
+                    "All remaining partner_id values should be unique.",
+                ],
+            },
+            "initial_table": [
+                {"row_id": 40, "partner_id": "P100", "name": "delta analytics", "city": "san francisco", "email": "hello@delta.example.com"},
+                {"row_id": 41, "partner_id": "P101", "name": "LUCIA ROMERO", "city": "MADRID", "email": "lucia.romero at example.com"},
+                {"row_id": 42, "partner_id": "P102", "name": "Ken Ito", "city": "Tokyo", "email": "ken.ito@example.com"},
+                {"row_id": 43, "partner_id": "P101", "name": "LUCIA ROMERO", "city": "MADRID", "email": "lucia.romero at example.com"},
+                {"row_id": 44, "partner_id": "P103", "name": "amina ali", "city": "dubai", "email": "amina.ali.example.com"},
+                {"row_id": 45, "partner_id": "P104", "name": "Sofia Hart", "city": "London", "email": "sofia.hart@example.com"},
+            ],
+            "hidden_issues": [
+                {
+                    "type": "duplicate",
+                    "rows": [41, 43],
+                    "description": "Rows 41 and 43 are duplicates and only one should remain.",
+                },
+                {
+                    "type": "inconsistent_casing",
+                    "rows": [40, 41, 44],
+                    "column": "name",
+                    "description": "Rows 40, 41, and 44 contain inconsistent casing in names.",
+                },
+                {
+                    "type": "inconsistent_casing",
+                    "rows": [40, 41, 44],
+                    "column": "city",
+                    "description": "Rows 40, 41, and 44 contain inconsistent casing in cities.",
+                },
+                {
+                    "type": "invalid_format",
+                    "row": 41,
+                    "column": "email",
+                    "description": "Row 41 contains an invalid email format.",
+                },
+                {
+                    "type": "invalid_format",
+                    "row": 44,
+                    "column": "email",
+                    "description": "Row 44 contains an invalid email format.",
+                },
+            ],
+            "constraints": [
+                "Preserve the original entity identity of each remaining row.",
+                "Normalize names and cities to a consistent human-readable casing style.",
+                "Only repair emails that are actually invalid.",
+                "Do not introduce new duplicates while normalizing values.",
+            ],
+            "max_steps": 9,
+        },
+    ]
+    return _pick_variant(variant, variants)
+
+
+def hard_conflict_resolution_task(variant: int | None = None) -> TaskDefinition:
+    """Create a hard multi-step conflict-resolution task with deceptive records."""
+
+    variants: List[TaskDefinition] = [
+        {
+            "goal": "Resolve conflicting records, enforce unique email constraints, fix invalid formats, and preserve valid but unusual data.",
+            "difficulty": "hard",
+            "variant_id": "hard_customer_conflicts",
+            "required_columns": ["name", "email", "phone", "status"],
+            "expected_outcome": {
+                "expected_row_count_range": {"min": 5, "max": 6},
+                "unique_by": ["email"],
+                "format_rules": {"email": "valid_email", "phone": "normalized_phone"},
+                "exactly_one_of_rows": [[21, 22], [23, 24], [26, 27]],
+                "must_preserve_valid_rows": [25, 28],
+                "validation_rules": [
+                    "Exactly one of rows 21 or 22 should remain after deduplication.",
+                    "Exactly one of rows 23 or 24 should remain after conflict resolution.",
+                    "Exactly one of rows 26 or 27 should remain after enforcing email uniqueness.",
+                    "No two remaining rows should share the same email address.",
+                    "All remaining emails should satisfy a valid email format.",
+                    "All remaining phone values should be normalized to a consistent valid format.",
+                    "Rows 25 and 28 should remain logically unchanged because they are valid trap rows.",
+                ],
+            },
+            "initial_table": [
+                {"row_id": 21, "customer_id": "C200", "name": "Nina Patel", "email": "nina.patel@example.com", "phone": "206-555-0101", "status": "active"},
+                {"row_id": 22, "customer_id": "C200", "name": "Nina Patel", "email": "nina.patel@example.com", "phone": "206-555-0101", "status": "active"},
+                {"row_id": 23, "customer_id": "C201", "name": "Evan Cole", "email": "evan.cole@example", "phone": "4155550102", "status": "active"},
+                {"row_id": 24, "customer_id": "C201", "name": "Evan Cole", "email": "evan.cole@example.com", "phone": "(415) 555-0102", "status": "inactive"},
+                {"row_id": 25, "customer_id": "C202", "name": "A. J. Brown", "email": "aj.brown@example.com", "phone": "+1-312-555-0103", "status": "active"},
+                {"row_id": 26, "customer_id": "C203", "name": "Marta Silva", "email": "shared@example.com", "phone": "646-555-0104", "status": "active"},
+                {"row_id": 27, "customer_id": "C204", "name": "Martin Silva", "email": "shared@example.com", "phone": "646-555-0105", "status": "active"},
+                {"row_id": 28, "customer_id": "C205", "name": "Q Xu", "email": "q.xu+vip@example.com", "phone": "917-555-0106", "status": "active"},
+            ],
+            "hidden_issues": [
+                {
+                    "type": "duplicate",
+                    "rows": [21, 22],
+                    "description": "Rows 21 and 22 are exact duplicates and only one should remain.",
+                },
+                {
+                    "type": "conflict",
+                    "rows": [23, 24],
+                    "description": "Rows 23 and 24 conflict for the same customer and must be reconciled into one trustworthy record.",
+                },
+                {
+                    "type": "invalid_format",
+                    "row": 23,
+                    "column": "email",
+                    "description": "Row 23 contains an invalid email format.",
+                },
+                {
+                    "type": "invalid_format",
+                    "row": 23,
+                    "column": "phone",
+                    "description": "Row 23 contains an invalid phone format.",
+                },
+                {
+                    "type": "constraint_violation",
+                    "constraint": "unique_email",
+                    "rows": [26, 27],
+                    "description": "Rows 26 and 27 violate the unique email constraint.",
+                },
+                {
+                    "type": "valid_trap",
+                    "row": 28,
+                    "description": "Row 28 is valid even though the plus-address format may look suspicious.",
+                },
+                {
+                    "type": "valid_trap",
+                    "row": 25,
+                    "description": "Row 25 is valid even though the name abbreviation may look inconsistent.",
+                },
+            ],
+            "constraints": [
+                "Email values must be unique across the final table.",
+                "Every remaining row must represent a single coherent customer record.",
+                "Do not modify valid rows just because they look unusual.",
+                "Prefer correction and conflict resolution over unnecessary deletion.",
+            ],
+            "max_steps": 14,
+        },
+        {
+            "goal": "Resolve conflicting records, enforce unique email constraints, fix invalid formats, and preserve valid but unusual data.",
+            "difficulty": "hard",
+            "variant_id": "hard_account_merges",
+            "required_columns": ["name", "email", "phone", "status"],
+            "expected_outcome": {
+                "expected_row_count_range": {"min": 5, "max": 6},
+                "unique_by": ["email"],
+                "format_rules": {"email": "valid_email", "phone": "normalized_phone"},
+                "exactly_one_of_rows": [[51, 52], [53, 54], [56, 57]],
+                "must_preserve_valid_rows": [55, 58],
+                "validation_rules": [
+                    "Exactly one of rows 51 or 52 should remain after deduplication.",
+                    "Exactly one of rows 53 or 54 should remain after conflict resolution.",
+                    "Exactly one of rows 56 or 57 should remain after enforcing email uniqueness.",
+                    "No two remaining rows should share the same email address.",
+                    "All remaining emails should satisfy a valid email format.",
+                    "All remaining phone values should be normalized to a consistent valid format.",
+                    "Rows 55 and 58 should remain logically unchanged because they are valid trap rows.",
+                ],
+            },
+            "initial_table": [
+                {"row_id": 51, "customer_id": "A900", "name": "Lena Brooks", "email": "lena.brooks@example.com", "phone": "212-555-0111", "status": "active"},
+                {"row_id": 52, "customer_id": "A900", "name": "Lena Brooks", "email": "lena.brooks@example.com", "phone": "212-555-0111", "status": "active"},
+                {"row_id": 53, "customer_id": "A901", "name": "Ravi Shah", "email": "ravi.shah example.com", "phone": "6465550112", "status": "active"},
+                {"row_id": 54, "customer_id": "A901", "name": "Ravi Shah", "email": "ravi.shah@example.com", "phone": "646-555-0112", "status": "inactive"},
+                {"row_id": 55, "customer_id": "A902", "name": "M. E. Klein", "email": "mek@example.com", "phone": "+1-303-555-0113", "status": "active"},
+                {"row_id": 56, "customer_id": "A903", "name": "Sana Noor", "email": "ops@example.com", "phone": "718-555-0114", "status": "active"},
+                {"row_id": 57, "customer_id": "A904", "name": "Sana N.", "email": "ops@example.com", "phone": "718-555-0115", "status": "active"},
+                {"row_id": 58, "customer_id": "A905", "name": "Bo Li", "email": "bo.li+archive@example.com", "phone": "415-555-0116", "status": "active"},
+            ],
+            "hidden_issues": [
+                {
+                    "type": "duplicate",
+                    "rows": [51, 52],
+                    "description": "Rows 51 and 52 are exact duplicates and only one should remain.",
+                },
+                {
+                    "type": "conflict",
+                    "rows": [53, 54],
+                    "description": "Rows 53 and 54 conflict for the same customer and must be reconciled into one trustworthy record.",
+                },
+                {
+                    "type": "invalid_format",
+                    "row": 53,
+                    "column": "email",
+                    "description": "Row 53 contains an invalid email format.",
+                },
+                {
+                    "type": "invalid_format",
+                    "row": 53,
+                    "column": "phone",
+                    "description": "Row 53 contains an invalid phone format.",
+                },
+                {
+                    "type": "constraint_violation",
+                    "constraint": "unique_email",
+                    "rows": [56, 57],
+                    "description": "Rows 56 and 57 violate the unique email constraint.",
+                },
+                {
+                    "type": "valid_trap",
+                    "row": 55,
+                    "description": "Row 55 is valid even though the abbreviated name may look unusual.",
+                },
+                {
+                    "type": "valid_trap",
+                    "row": 58,
+                    "description": "Row 58 is valid even though the plus-address format may look suspicious.",
+                },
+            ],
+            "constraints": [
+                "Email values must be unique across the final table.",
+                "Every remaining row must represent a single coherent customer record.",
+                "Do not modify valid rows just because they look unusual.",
+                "Prefer correction and conflict resolution over unnecessary deletion.",
+            ],
+            "max_steps": 14,
+        },
+    ]
+    return _pick_variant(variant, variants)
+
+
+easy_cleaning_task.variant_count = 2
+medium_normalization_task.variant_count = 2
+hard_conflict_resolution_task.variant_count = 2

utils/__init__.py

@@ -0,0 +1 @@
+"""Utility package for the dataops OpenEnv environment."""

utils/helpers.py

@@ -0,0 +1,11 @@
+"""Utility helpers for ``dataops-gym``.
+
+This module is responsible for small shared helper functions that support the
+environment without owning core business or orchestration logic.
+"""
+
+from __future__ import annotations
+
+
+# TODO: Add reusable helper utilities with clear, narrow responsibilities.
+# TODO: Avoid placing core domain logic in shared helpers.