Fix critical issues from code review

- Dockerfile: add non-root user (required by HF Spaces)
- app.py: add asyncio lock for concurrent request safety
- environment.py: fix score inconsistency — use _recompute_score()
consistently instead of ad-hoc per-step score manipulation
- environment.py: fix validate_row_deleted false positive — use index
tracking instead of content matching for duplicate detection
- client.py: fix step payload to wrap action in {"action": {...}}
- Remove duplicate data_clean_env/inference.py (only root one needed)

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Dockerfile

@@ -7,9 +7,6 @@ RUN apt-get update && \
     apt-get install -y --no-install-recommends git curl && \
     rm -rf /var/lib/apt/lists/*
 
-# Copy everything
-COPY . /app
-
 # Install Python dependencies
 RUN pip install --no-cache-dir \
     "openenv-core>=0.2.3" \
@@ -19,7 +16,14 @@ RUN pip install --no-cache-dir \
     "requests>=2.31.0" \
     "openai>=1.0.0"
 
-ENV PYTHONPATH="/app:$PYTHONPATH"
+# Copy everything
+COPY . /app
+
+# Create non-root user (required by HF Spaces)
+RUN useradd -m -u 1000 user && chown -R user:user /app
+USER user
+
+ENV PYTHONPATH="/app"
 
 HEALTHCHECK --interval=30s --timeout=3s --start-period=10s --retries=3 \
     CMD python -c "import urllib.request; urllib.request.urlopen('http://localhost:8000/health')" || exit 1

data_clean_env/client.py

@@ -20,7 +20,7 @@ class DataCleanEnv(EnvClient[DataCleanAction, DataCleanObservation, DataCleanSta
     """
 
     def _step_payload(self, action: DataCleanAction) -> dict:
-        return action.model_dump()
+        return {"action": action.model_dump()}
 
     def _parse_result(self, payload: dict) -> StepResult[DataCleanObservation]:
         obs = DataCleanObservation(**payload.get("observation", payload))

data_clean_env/inference.py

@@ -1,284 +0,0 @@
-"""
-Inference Script — DataCleanEnv
-===================================
-MANDATORY
-- Before submitting, ensure the following variables are defined in your environment configuration:
-    API_BASE_URL   The API endpoint for the LLM.
-    MODEL_NAME     The model identifier to use for inference.
-    HF_TOKEN       Your Hugging Face / API key.
-
-- The inference script must be named `inference.py` and placed in the root directory of the project
-- Participants must use OpenAI Client for all LLM calls using above variables
-"""
-
-import json
-import os
-import re
-import sys
-import textwrap
-from typing import Any, Dict, List
-
-from openai import OpenAI
-
-# ---------------------------------------------------------------------------
-# Config from environment
-# ---------------------------------------------------------------------------
-API_BASE_URL = os.getenv("API_BASE_URL", "https://router.huggingface.co/v1")
-API_KEY = os.getenv("HF_TOKEN") or os.getenv("API_KEY", "")
-MODEL_NAME = os.getenv("MODEL_NAME", "")
-ENV_URL = os.getenv("ENV_URL", "http://localhost:8000")
-
-TEMPERATURE = 0.0
-MAX_TOKENS = 300
-
-TASKS = ["customer_contacts", "sales_records", "employee_records"]
-
-# ---------------------------------------------------------------------------
-# System prompt
-# ---------------------------------------------------------------------------
-SYSTEM_PROMPT = textwrap.dedent("""\
-    You are an expert data quality analyst. You are given a dataset with known issues.
-    Your job is to find and fix ALL data quality problems efficiently.
-
-    Available commands (respond with EXACTLY ONE command per turn, no explanation):
-      inspect("column_name")                — View column statistics and issues
-      fix(row_index, "column_name", "value") — Fix a specific cell value
-      delete(row_index)                      — Remove a duplicate/invalid row
-      submit()                               — Finalize and get your score
-
-    STRATEGY (follow this order strictly):
-    1. You will first receive inspection results for all columns — read them carefully
-    2. Look at the data table and identify ALL issues based on the inspection hints
-    3. Fix ALL issues using fix() commands, one at a time
-    4. Delete duplicate rows LAST (after all fixes), from highest row index to lowest
-       (this avoids row index shifting problems)
-    5. Only call submit() when you believe ALL issues are fixed
-
-    VALIDATION RULES:
-    - Emails: must match user@domain.tld (no [at], no @@, no spaces)
-    - Phone numbers: digits, dashes, parens, spaces only; at least 10 digits
-    - Dates: YYYY-MM-DD format (e.g., 2024-03-25), must be valid calendar date
-    - Empty/missing values: provide a reasonable non-empty value matching context
-    - Negative numbers (quantity/price): make them positive (use absolute value)
-    - Price/salary outliers: fix to a reasonable value within the stated range
-    - Inconsistent region/department names: use the EXACT canonical form from the task description
-    - Excess whitespace: trim leading/trailing spaces, collapse double spaces to single
-    - Manager IDs: must reference an existing employee emp_id in the dataset
-    - Performance scores: must be within 0.0-10.0
-    - Salaries: must be within $20,000-$500,000
-    - Termination dates: must be AFTER hire date, or leave empty if employee is active
-    - Duplicate rows: rows that are exact or near-exact copies — delete the LATER one
-
-    IMPORTANT: After deleting a row, all subsequent row indices shift down by 1.
-    Always delete from highest index to lowest to avoid this issue.
-
-    Respond with ONLY the command. No explanation, no markdown, no extra text.
-""")
-
-
-# ---------------------------------------------------------------------------
-# HTTP helpers (direct REST calls to the environment server)
-# ---------------------------------------------------------------------------
-import requests
-
-
-def env_reset(task_id: str) -> Dict[str, Any]:
-    """Reset the environment with a specific task."""
-    resp = requests.post(
-        f"{ENV_URL}/reset",
-        json={"task_id": task_id},
-        timeout=30,
-    )
-    resp.raise_for_status()
-    return resp.json()
-
-
-def env_step(command: str) -> Dict[str, Any]:
-    """Execute a command in the environment."""
-    resp = requests.post(
-        f"{ENV_URL}/step",
-        json={"action": {"command": command}},
-        timeout=30,
-    )
-    resp.raise_for_status()
-    return resp.json()
-
-
-# ---------------------------------------------------------------------------
-# Observation formatting
-# ---------------------------------------------------------------------------
-def format_observation(obs: Dict[str, Any], step_num: int) -> str:
-    """Format an observation dict as a user message for the LLM."""
-    parts = []
-    parts.append(f"Step {step_num} | Task: {obs.get('task_id', '?')} ({obs.get('difficulty', '?')})")
-    parts.append(f"Issues fixed: {obs.get('issues_fixed', 0)}/{obs.get('total_issues', 0)} | Score: {obs.get('current_score', 0.0):.4f}")
-    parts.append(f"Actions remaining: {obs.get('actions_remaining', 0)}")
-    parts.append("")
-
-    feedback = obs.get("feedback", "")
-    if feedback:
-        parts.append(f"Last action result: {feedback}")
-        parts.append("")
-
-    parts.append("Task description:")
-    parts.append(obs.get("task_description", ""))
-    parts.append("")
-
-    parts.append("Column info:")
-    parts.append(obs.get("column_info", ""))
-    parts.append("")
-
-    parts.append("Current data:")
-    parts.append(obs.get("data_preview", ""))
-
-    return "\n".join(parts)
-
-
-# ---------------------------------------------------------------------------
-# Action extraction
-# ---------------------------------------------------------------------------
-ACTION_RE = re.compile(r"(inspect|fix|delete|submit)\s*\(", re.IGNORECASE)
-
-
-def extract_action(response_text: str) -> str:
-    """Extract a valid action from the LLM response."""
-    if not response_text:
-        return "submit()"
-
-    # Try each line
-    for line in response_text.strip().splitlines():
-        line = line.strip()
-        if not line:
-            continue
-        # Strip markdown code fences
-        line = re.sub(r"^```\w*\s*", "", line)
-        line = re.sub(r"\s*```$", "", line)
-        # Strip "action:" prefix
-        line = re.sub(r"^(?:action|next action)\s*[:\-]\s*", "", line, flags=re.IGNORECASE)
-        if ACTION_RE.search(line):
-            m = ACTION_RE.search(line)
-            start = m.start()
-            depth = 0
-            for i in range(start, len(line)):
-                if line[i] == "(":
-                    depth += 1
-                elif line[i] == ")":
-                    depth -= 1
-                    if depth == 0:
-                        return line[start : i + 1]
-            return line[start:] + ")"
-
-    return "submit()"
-
-
-# ---------------------------------------------------------------------------
-# Main
-# ---------------------------------------------------------------------------
-def main() -> None:
-    client = OpenAI(base_url=API_BASE_URL, api_key=API_KEY)
-    results = {}
-
-    for task_id in TASKS:
-        print(f"\n{'=' * 60}")
-        print(f"Task: {task_id}")
-        print(f"{'=' * 60}")
-
-        obs = env_reset(task_id)
-        if "observation" in obs:
-            obs = obs["observation"]
-
-        done = obs.get("done", False)
-        messages = [{"role": "system", "content": SYSTEM_PROMPT}]
-
-        # --- Phase 1: Auto-inspect all columns ---
-        columns = []
-        col_info = obs.get("column_info", "")
-        for line in col_info.strip().splitlines():
-            line = line.strip()
-            if ":" in line:
-                col_name = line.split(":")[0].strip()
-                if col_name:
-                    columns.append(col_name)
-
-        inspection_results = []
-        step_num = 0
-        for col in columns:
-            if done:
-                break
-            step_num += 1
-            cmd = f'inspect("{col}")'
-            print(f"  Step {step_num}: {cmd}")
-            obs = env_step(cmd)
-            if "observation" in obs:
-                obs = obs["observation"]
-            done = obs.get("done", False)
-            feedback = obs.get("feedback", "")
-            inspection_results.append(f"[{col}]: {feedback}")
-
-        if done:
-            score = obs.get("current_score", 0.0)
-            print(f"  Done during inspection! Score: {score:.4f}")
-            results[task_id] = score
-            continue
-
-        # Build a summary of all inspections for the LLM
-        inspection_summary = "\n\n".join(inspection_results)
-        initial_context = format_observation(obs, step_num)
-        initial_context += f"\n\n--- INSPECTION SUMMARY (all columns) ---\n{inspection_summary}"
-        initial_context += "\n\n--- NOW FIX ALL ISSUES. Do fixes first, then deletions (highest index first). ---"
-
-        messages.append({"role": "user", "content": initial_context})
-
-        # --- Phase 2: LLM-driven fixes ---
-        while not done:
-            step_num += 1
-
-            # Keep conversation manageable — but preserve system + initial context
-            if len(messages) > 40:
-                messages = [messages[0], messages[1]] + messages[-36:]
-
-            try:
-                completion = client.chat.completions.create(
-                    model=MODEL_NAME,
-                    messages=messages,
-                    temperature=TEMPERATURE,
-                    max_tokens=MAX_TOKENS,
-                    stream=False,
-                )
-                response_text = completion.choices[0].message.content or ""
-            except Exception as exc:
-                print(f"  LLM error: {exc}. Using submit().")
-                response_text = "submit()"
-
-            action = extract_action(response_text)
-            messages.append({"role": "assistant", "content": action})
-            print(f"  Step {step_num}: {action}")
-
-            obs = env_step(action)
-            if "observation" in obs:
-                obs = obs["observation"]
-
-            done = obs.get("done", False)
-            score = obs.get("current_score", 0.0)
-
-            if done:
-                print(f"  Done! Score: {score:.4f}")
-                results[task_id] = score
-            else:
-                # Send updated observation as next user message
-                user_msg = format_observation(obs, step_num)
-                messages.append({"role": "user", "content": user_msg})
-
-        print(f"  Final score for {task_id}: {results.get(task_id, 0.0):.4f}")
-
-    print(f"\n{'=' * 60}")
-    print("RESULTS SUMMARY")
-    print(f"{'=' * 60}")
-    for task_id, score in results.items():
-        print(f"  {task_id}: {score:.4f}")
-    avg = sum(results.values()) / len(results) if results else 0.0
-    print(f"  Average: {avg:.4f}")
-
-
-if __name__ == "__main__":
-    main()

data_clean_env/server/app.py

@@ -1,9 +1,10 @@
 """FastAPI application for the DataClean Environment."""
 
+import asyncio
+
 from fastapi import FastAPI, Body
-from fastapi.responses import JSONResponse
 from pydantic import BaseModel
-from typing import Any, Dict, List, Optional
+from typing import Any, Dict, Optional
 
 try:
     from .environment import DataCleanEnvironment
@@ -16,8 +17,9 @@ except ImportError:
 
 app = FastAPI(title="DataCleanEnv", version="1.0.0")
 
-# Shared stateful environment instance
+# Shared stateful environment instance with lock for concurrent safety
 _env = DataCleanEnvironment()
+_env_lock = asyncio.Lock()
 
 
 # ---------------------------------------------------------------------------
@@ -49,14 +51,16 @@ async def health():
 
 @app.post("/reset")
 async def reset(request: ResetRequest = Body(default_factory=ResetRequest)):
-    obs = _env.reset(seed=request.seed, episode_id=request.episode_id, task_id=request.task_id)
+    async with _env_lock:
+        obs = _env.reset(seed=request.seed, episode_id=request.episode_id, task_id=request.task_id)
     return {"observation": _obs_dict(obs), "reward": None, "done": False}
 
 
 @app.post("/step")
 async def step(request: StepRequest):
     action = DataCleanAction(**request.action)
-    obs = _env.step(action)
+    async with _env_lock:
+        obs = _env.step(action)
     return {"observation": _obs_dict(obs), "reward": obs.reward, "done": obs.done}
 
 

data_clean_env/server/environment.py

@@ -288,7 +288,7 @@ class DataCleanEnvironment(Environment):
             # Agent is modifying a cell that had no known issue — damage
             self._current_data[row_idx][col] = new_value
             self._damaged_cells += 1
-            self._score = max(0.0, self._score - 0.05)
+            self._recompute_score()
             return (
                 f"Warning: Row {row_idx}, column '{col}' had no known issue. "
                 f"Original value '{old_value}' was overwritten. Penalty applied (-0.05)."
@@ -313,7 +313,7 @@ class DataCleanEnvironment(Environment):
                 reward_delta += 1.0 / len(self._task.issues)
                 fixed_any = True
 
-        self._score = min(1.0, max(0.0, self._score + reward_delta))
+        self._recompute_score()
 
         if fixed_any:
             return (
@@ -337,7 +337,6 @@ class DataCleanEnvironment(Environment):
         deleted_row = self._current_data[row_idx]
 
         # Check if this row is a known duplicate
-        reward_delta = 0.0
         is_duplicate = False
         for issue in self._task.issues:
             if issue.issue_type != "duplicate_row":
@@ -347,21 +346,18 @@ class DataCleanEnvironment(Environment):
             if self._issue_status.get(issue.issue_id, False):
                 continue
 
-            # The agent is deleting the row that matches the duplicate issue
             self._issue_status[issue.issue_id] = True
-            reward_delta += 1.0 / len(self._task.issues)
             is_duplicate = True
 
         if not is_duplicate:
             # Deleting a non-duplicate row — penalty
             self._damaged_cells += 1
-            reward_delta = -0.05
 
         # Actually remove the row
         self._current_data.pop(row_idx)
         self._row_index_map.pop(row_idx)
 
-        self._score = min(1.0, max(0.0, self._score + reward_delta))
+        self._recompute_score()
 
         if is_duplicate:
             return (
@@ -380,7 +376,8 @@ class DataCleanEnvironment(Environment):
     def _check_issue_resolved(self, issue: Issue) -> bool:
         """Check if an issue is resolved in the current data."""
         if issue.issue_type == "duplicate_row":
-            return validate_row_deleted(self._current_data, issue.original_row_data)
+            # Row is resolved if it was deleted (no longer in index map)
+            return self._find_current_row(issue.row) is None
 
         if issue.issue_type == "temporal_inconsistency":
             # Find the row by original index
@@ -425,6 +422,17 @@ class DataCleanEnvironment(Environment):
                 return new_value
         return new_value
 
+    def _recompute_score(self) -> None:
+        """Recompute running score based on current issue states and damage."""
+        total = len(self._task.issues)
+        if total == 0:
+            self._score = 1.0
+            return
+        resolved = sum(1 for v in self._issue_status.values() if v)
+        self._score = resolved / total
+        self._score = max(0.0, self._score - self._damaged_cells * 0.05)
+        self._score = min(1.0, self._score)
+
     def _compute_final_score(self) -> None:
         """Recompute score based on all current issue states."""
         total = len(self._task.issues)