envrionment setup

.gitignore

@@ -0,0 +1,439 @@
+# Byte-compiled / optimized / DLL files
+__pycache__/
+__pycache__/__init.cpython-313.pyc
+__pycache__/client.cpython-313.pyc
+# Ignore changes to __pycache__
+__pycache__/
+*.py[codz]
+*$py.class
+
+# C extensions
+*.so
+
+# Distribution / packaging
+.Python
+build/
+develop-eggs/
+dist/
+downloads/
+eggs/
+.eggs/
+lib/
+lib64/
+parts/
+sdist/
+var/
+wheels/
+share/python-wheels/
+*.egg-info/
+.installed.cfg
+*.egg
+MANIFEST
+
+# PyInstaller
+#   Usually these files are written by a python script from a template
+#   before PyInstaller builds the exe, so as to inject date/other infos into it.
+*.manifest
+*.spec
+
+# Installer logs
+pip-log.txt
+pip-delete-this-directory.txt
+
+# Unit test / coverage reports
+htmlcov/
+.tox/
+.nox/
+.coverage
+.coverage.*
+.cache
+nosetests.xml
+coverage.xml
+*.cover
+*.py.cover
+.hypothesis/
+.pytest_cache/
+cover/
+
+# Translations
+*.mo
+*.pot
+
+# Django stuff:
+*.log
+local_settings.py
+db.sqlite3
+db.sqlite3-journal
+
+# Flask stuff:
+instance/
+.webassets-cache
+
+# Scrapy stuff:
+.scrapy
+
+# Sphinx documentation
+docs/_build/
+
+# PyBuilder
+.pybuilder/
+target/
+
+# Jupyter Notebook
+.ipynb_checkpoints
+
+# IPython
+profile_default/
+ipython_config.py
+
+# pyenv
+#   For a library or package, you might want to ignore these files since the code is
+#   intended to run in multiple environments; otherwise, check them in:
+# .python-version
+
+# pipenv
+#   According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
+#   However, in case of collaboration, if having platform-specific dependencies or dependencies
+#   having no cross-platform support, pipenv may install dependencies that don't work, or not
+#   install all needed dependencies.
+# Pipfile.lock
+
+# UV
+#   Similar to Pipfile.lock, it is generally recommended to include uv.lock in version control.
+#   This is especially recommended for binary packages to ensure reproducibility, and is more
+#   commonly ignored for libraries.
+# uv.lock
+
+# poetry
+#   Similar to Pipfile.lock, it is generally recommended to include poetry.lock in version control.
+#   This is especially recommended for binary packages to ensure reproducibility, and is more
+#   commonly ignored for libraries.
+#   https://python-poetry.org/docs/basic-usage/#commit-your-poetrylock-file-to-version-control
+# poetry.lock
+# poetry.toml
+
+# pdm
+#   Similar to Pipfile.lock, it is generally recommended to include pdm.lock in version control.
+#   pdm recommends including project-wide configuration in pdm.toml, but excluding .pdm-python.
+#   https://pdm-project.org/en/latest/usage/project/#working-with-version-control
+# pdm.lock
+# pdm.toml
+.pdm-python
+.pdm-build/
+
+# pixi
+#   Similar to Pipfile.lock, it is generally recommended to include pixi.lock in version control.
+# pixi.lock
+#   Pixi creates a virtual environment in the .pixi directory, just like venv module creates one
+#   in the .venv directory. It is recommended not to include this directory in version control.
+.pixi
+
+# PEP 582; used by e.g. github.com/David-OConnor/pyflow and github.com/pdm-project/pdm
+__pypackages__/
+
+# Celery stuff
+celerybeat-schedule
+celerybeat.pid
+
+# Redis
+*.rdb
+*.aof
+*.pid
+
+# RabbitMQ
+mnesia/
+rabbitmq/
+rabbitmq-data/
+
+# ActiveMQ
+activemq-data/
+
+# SageMath parsed files
+*.sage.py
+
+# Environments
+.env
+.envrc
+.venv
+env/
+venv/
+ENV/
+env.bak/
+venv.bak/
+
+# Spyder project settings
+.spyderproject
+.spyproject
+
+# Rope project settings
+.ropeproject
+
+# mkdocs documentation
+/site
+
+# mypy
+.mypy_cache/
+.dmypy.json
+dmypy.json
+
+# Pyre type checker
+.pyre/
+
+# pytype static type analyzer
+.pytype/
+
+# Cython debug symbols
+cython_debug/
+
+# PyCharm
+#   JetBrains specific template is maintained in a separate JetBrains.gitignore that can
+#   be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore
+#   and can be added to the global gitignore or merged into this file.  For a more nuclear
+#   option (not recommended) you can uncomment the following to ignore the entire idea folder.
+# .idea/
+.env# Byte-compiled / optimized / DLL files
+__pycache__/
+__pycache__/__init.cpython-313.pyc
+__pycache__/client.cpython-313.pyc
+# Ignore changes to __pycache__
+__pycache__/
+*.py[codz]
+*$py.class
+
+# C extensions
+*.so
+
+# Distribution / packaging
+.Python
+build/
+develop-eggs/
+dist/
+downloads/
+eggs/
+.eggs/
+lib/
+lib64/
+parts/
+sdist/
+var/
+wheels/
+share/python-wheels/
+*.egg-info/
+.installed.cfg
+*.egg
+MANIFEST
+
+# PyInstaller
+#   Usually these files are written by a python script from a template
+#   before PyInstaller builds the exe, so as to inject date/other infos into it.
+*.manifest
+*.spec
+
+# Installer logs
+pip-log.txt
+pip-delete-this-directory.txt
+
+# Unit test / coverage reports
+htmlcov/
+.tox/
+.nox/
+.coverage
+.coverage.*
+.cache
+nosetests.xml
+coverage.xml
+*.cover
+*.py.cover
+.hypothesis/
+.pytest_cache/
+cover/
+
+# Translations
+*.mo
+*.pot
+
+# Django stuff:
+*.log
+local_settings.py
+db.sqlite3
+db.sqlite3-journal
+
+# Flask stuff:
+instance/
+.webassets-cache
+
+# Scrapy stuff:
+.scrapy
+
+# Sphinx documentation
+docs/_build/
+
+# PyBuilder
+.pybuilder/
+target/
+
+# Jupyter Notebook
+.ipynb_checkpoints
+
+# IPython
+profile_default/
+ipython_config.py
+
+# pyenv
+#   For a library or package, you might want to ignore these files since the code is
+#   intended to run in multiple environments; otherwise, check them in:
+# .python-version
+
+# pipenv
+#   According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
+#   However, in case of collaboration, if having platform-specific dependencies or dependencies
+#   having no cross-platform support, pipenv may install dependencies that don't work, or not
+#   install all needed dependencies.
+# Pipfile.lock
+
+# UV
+#   Similar to Pipfile.lock, it is generally recommended to include uv.lock in version control.
+#   This is especially recommended for binary packages to ensure reproducibility, and is more
+#   commonly ignored for libraries.
+# uv.lock
+
+# poetry
+#   Similar to Pipfile.lock, it is generally recommended to include poetry.lock in version control.
+#   This is especially recommended for binary packages to ensure reproducibility, and is more
+#   commonly ignored for libraries.
+#   https://python-poetry.org/docs/basic-usage/#commit-your-poetrylock-file-to-version-control
+# poetry.lock
+# poetry.toml
+
+# pdm
+#   Similar to Pipfile.lock, it is generally recommended to include pdm.lock in version control.
+#   pdm recommends including project-wide configuration in pdm.toml, but excluding .pdm-python.
+#   https://pdm-project.org/en/latest/usage/project/#working-with-version-control
+# pdm.lock
+# pdm.toml
+.pdm-python
+.pdm-build/
+
+# pixi
+#   Similar to Pipfile.lock, it is generally recommended to include pixi.lock in version control.
+# pixi.lock
+#   Pixi creates a virtual environment in the .pixi directory, just like venv module creates one
+#   in the .venv directory. It is recommended not to include this directory in version control.
+.pixi
+
+# PEP 582; used by e.g. github.com/David-OConnor/pyflow and github.com/pdm-project/pdm
+__pypackages__/
+
+# Celery stuff
+celerybeat-schedule
+celerybeat.pid
+
+# Redis
+*.rdb
+*.aof
+*.pid
+
+# RabbitMQ
+mnesia/
+rabbitmq/
+rabbitmq-data/
+
+# ActiveMQ
+activemq-data/
+
+# SageMath parsed files
+*.sage.py
+
+# Environments
+.env
+.envrc
+.venv
+env/
+venv/
+ENV/
+env.bak/
+venv.bak/
+
+# Spyder project settings
+.spyderproject
+.spyproject
+
+# Rope project settings
+.ropeproject
+
+# mkdocs documentation
+/site
+
+# mypy
+.mypy_cache/
+.dmypy.json
+dmypy.json
+
+# Pyre type checker
+.pyre/
+
+# pytype static type analyzer
+.pytype/
+
+# Cython debug symbols
+cython_debug/
+
+# PyCharm
+#   JetBrains specific template is maintained in a separate JetBrains.gitignore that can
+#   be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore
+#   and can be added to the global gitignore or merged into this file.  For a more nuclear
+#   option (not recommended) you can uncomment the following to ignore the entire idea folder.
+# .idea/
+
+# Abstra
+#   Abstra is an AI-powered process automation framework.
+#   Ignore directories containing user credentials, local state, and settings.
+#   Learn more at https://abstra.io/docs
+.abstra/
+
+# Visual Studio Code
+#   Visual Studio Code specific template is maintained in a separate VisualStudioCode.gitignore 
+#   that can be found at https://github.com/github/gitignore/blob/main/Global/VisualStudioCode.gitignore
+#   and can be added to the global gitignore or merged into this file. However, if you prefer, 
+#   you could uncomment the following to ignore the entire vscode folder
+# .vscode/
+
+# Ruff stuff:
+.ruff_cache/
+
+# PyPI configuration file
+.pypirc
+
+# Marimo
+marimo/_static/
+marimo/_lsp/
+__marimo__/
+
+# Streamlit
+.streamlit/secrets.toml
+# Abstra
+#   Abstra is an AI-powered process automation framework.
+#   Ignore directories containing user credentials, local state, and settings.
+#   Learn more at https://abstra.io/docs
+.abstra/
+
+# Visual Studio Code
+#   Visual Studio Code specific template is maintained in a separate VisualStudioCode.gitignore 
+#   that can be found at https://github.com/github/gitignore/blob/main/Global/VisualStudioCode.gitignore
+#   and can be added to the global gitignore or merged into this file. However, if you prefer, 
+#   you could uncomment the following to ignore the entire vscode folder
+# .vscode/
+
+# Ruff stuff:
+.ruff_cache/
+
+# PyPI configuration file
+.pypirc
+
+# Marimo
+marimo/_static/
+marimo/_lsp/
+__marimo__/
+
+# Streamlit
+.streamlit/secrets.toml

.vscode/settings.json

@@ -0,0 +1,4 @@
+{
+    "python-envs.defaultEnvManager": "ms-python.python:venv",
+    "python-envs.defaultPackageManager": "ms-python.python:pip"
+}

__init__.py

@@ -1,16 +1,22 @@
-# Copyright (c) Meta Platforms, Inc. and affiliates.
-# All rights reserved.
-#
-# This source code is licensed under the BSD-style license found in the
-# LICENSE file in the root directory of this source tree.
+"""Public package exports for python_code_review_env."""
 
-"""Python Env Environment."""
-
-from .client import PythonEnv
-from .models import PythonAction, PythonObservation
+from .client import PythonCodeReviewEnv, PythonEnv
+from .models import (
+    PythonAction,
+    PythonCodeReviewAction,
+    PythonCodeReviewObservation,
+    PythonCodeReviewState,
+    PythonObservation,
+    PythonState,
+)
 
 __all__ = [
     "PythonAction",
     "PythonObservation",
+    "PythonState",
+    "PythonCodeReviewAction",
+    "PythonCodeReviewObservation",
+    "PythonCodeReviewState",
+    "PythonCodeReviewEnv",
     "PythonEnv",
 ]

client.py

@@ -1,99 +1,37 @@
-# Copyright (c) Meta Platforms, Inc. and affiliates.
-# All rights reserved.
-#
-# This source code is licensed under the BSD-style license found in the
-# LICENSE file in the root directory of this source tree.
+"""Client helpers for python_code_review_env."""
 
-"""Python Env Environment Client."""
+from __future__ import annotations
 
 from typing import Dict
 
 from openenv.core import EnvClient
 from openenv.core.client_types import StepResult
-from openenv.core.env_server.types import State
 
-from .models import PythonAction, PythonObservation
+from .models import (
+    PythonCodeReviewAction,
+    PythonCodeReviewObservation,
+    PythonCodeReviewState,
+)
 
 
-class PythonEnv(
-    EnvClient[PythonAction, PythonObservation, State]
+class PythonCodeReviewEnv(
+    EnvClient[PythonCodeReviewAction, PythonCodeReviewObservation, PythonCodeReviewState]
 ):
-    """
-    Client for the Python Env Environment.
+    """Typed client for the code review environment."""
 
-    This client maintains a persistent WebSocket connection to the environment server,
-    enabling efficient multi-step interactions with lower latency.
-    Each client instance has its own dedicated environment session on the server.
-
-    Example:
-        >>> # Connect to a running server
-        >>> with PythonEnv(base_url="http://localhost:8000") as client:
-        ...     result = client.reset()
-        ...     print(result.observation.echoed_message)
-        ...
-        ...     result = client.step(PythonAction(message="Hello!"))
-        ...     print(result.observation.echoed_message)
-
-    Example with Docker:
-        >>> # Automatically start container and connect
-        >>> client = PythonEnv.from_docker_image("python_env-env:latest")
-        >>> try:
-        ...     result = client.reset()
-        ...     result = client.step(PythonAction(message="Test"))
-        ... finally:
-        ...     client.close()
-    """
-
-    def _step_payload(self, action: PythonAction) -> Dict:
-        """
-        Convert PythonAction to JSON payload for step message.
-
-        Args:
-            action: PythonAction instance
-
-        Returns:
-            Dictionary representation suitable for JSON encoding
-        """
-        return {
-            "message": action.message,
-        }
-
-    def _parse_result(self, payload: Dict) -> StepResult[PythonObservation]:
-        """
-        Parse server response into StepResult[PythonObservation].
-
-        Args:
-            payload: JSON response data from server
-
-        Returns:
-            StepResult with PythonObservation
-        """
-        obs_data = payload.get("observation", {})
-        observation = PythonObservation(
-            echoed_message=obs_data.get("echoed_message", ""),
-            message_length=obs_data.get("message_length", 0),
-            done=payload.get("done", False),
-            reward=payload.get("reward"),
-            metadata=obs_data.get("metadata", {}),
-        )
+    def _step_payload(self, action: PythonCodeReviewAction) -> Dict:
+        return action.model_dump(exclude_none=True)
 
+    def _parse_result(self, payload: Dict) -> StepResult[PythonCodeReviewObservation]:
+        observation = PythonCodeReviewObservation.model_validate(payload.get("observation", {}))
         return StepResult(
             observation=observation,
             reward=payload.get("reward"),
-            done=payload.get("done", False),
+            done=payload.get("done", observation.done),
         )
 
-    def _parse_state(self, payload: Dict) -> State:
-        """
-        Parse server response into State object.
+    def _parse_state(self, payload: Dict) -> PythonCodeReviewState:
+        return PythonCodeReviewState.model_validate(payload)
 
-        Args:
-            payload: JSON response from state request
 
-        Returns:
-            State object with episode_id and step_count
-        """
-        return State(
-            episode_id=payload.get("episode_id"),
-            step_count=payload.get("step_count", 0),
-        )
+PythonEnv = PythonCodeReviewEnv

compat.py

@@ -0,0 +1,13 @@
+"""Compatibility helpers expected by validator-oriented scripts."""
+
+from __future__ import annotations
+
+
+def install_openenv_fastmcp_compat() -> None:
+    """Install runtime shims when needed.
+
+    The current environment does not require any monkey-patching, so this is a
+    deliberate no-op kept for validator compatibility.
+    """
+
+    return None

graders/__init__.py

@@ -0,0 +1,5 @@
+"""Deterministic graders for python_code_review_env."""
+
+from .dispatch import grade_task
+
+__all__ = ["grade_task"]

graders/bug_fix.py

@@ -0,0 +1,90 @@
+"""Bug-fix task grader."""
+
+from __future__ import annotations
+
+try:
+    from ..models import TaskGrade
+    from ..tasks.catalog import ReviewTask
+except ImportError:
+    from models import TaskGrade
+    from tasks.catalog import ReviewTask
+
+from .shared import base_grade, compile_code, execute_cases, quality_metrics, similarity_score, summarize_results
+
+
+def grade_bug_fix_task(
+    task: ReviewTask,
+    code: str,
+    *,
+    include_hidden: bool,
+    timeout_s: float = 2.0,
+) -> TaskGrade:
+    """Grade a bug-fix task against public or full test suites."""
+
+    compiled, compile_error = compile_code(code)
+    quality = quality_metrics(code, task.function_name)
+    details = {
+        "compile_error": compile_error,
+        "quality_notes": quality["quality_notes"],
+        "style_score": quality["style_score"],
+        "visibility": "full" if include_hidden else "public",
+    }
+
+    if not compiled:
+        partial = round(min(0.2, similarity_score(code, task.reference_code) * 0.2), 3)
+        details["test_results"] = []
+        details["test_summary"] = "Code does not compile."
+        return base_grade(
+            score=partial,
+            syntax_score=0.0,
+            tests_passed=0,
+            tests_total=len(task.public_cases) + (len(task.hidden_cases) if include_hidden else 0),
+            quality_score=0.0,
+            runtime_score=0.0,
+            timed_out=False,
+            details=details,
+        )
+
+    cases = task.public_cases + (task.hidden_cases if include_hidden else [])
+    result = execute_cases(code, task.function_name, cases, timeout_s=timeout_s)
+    if result.get("timed_out"):
+        details["test_results"] = []
+        details["test_summary"] = result["error"]
+        return base_grade(
+            score=0.0,
+            syntax_score=1.0,
+            tests_passed=0,
+            tests_total=len(cases),
+            quality_score=quality["score"],
+            runtime_score=0.0,
+            timed_out=True,
+            details=details,
+        )
+    if "error" in result:
+        details["test_results"] = []
+        details["test_summary"] = result["error"]
+        return base_grade(
+            score=0.0,
+            syntax_score=1.0,
+            tests_passed=0,
+            tests_total=len(cases),
+            quality_score=quality["score"],
+            runtime_score=0.0,
+            timed_out=False,
+            details=details,
+        )
+
+    data = result["data"]
+    pass_rate = data["passed"] / max(data["total"], 1)
+    details["test_results"] = data["results"]
+    details["test_summary"] = summarize_results("Test results", data["results"])
+    return base_grade(
+        score=pass_rate,
+        syntax_score=1.0,
+        tests_passed=data["passed"],
+        tests_total=data["total"],
+        quality_score=quality["score"],
+        runtime_score=0.0,
+        timed_out=False,
+        details=details,
+    )

graders/dispatch.py

@@ -0,0 +1,32 @@
+"""Task grader dispatch."""
+
+from __future__ import annotations
+
+try:
+    from ..models import TaskGrade
+    from ..tasks.catalog import ReviewTask
+except ImportError:
+    from models import TaskGrade
+    from tasks.catalog import ReviewTask
+
+from .bug_fix import grade_bug_fix_task
+from .optimization import grade_optimization_task
+from .syntax import grade_syntax_task
+
+
+def grade_task(
+    task: ReviewTask,
+    code: str,
+    *,
+    include_hidden: bool,
+    timeout_s: float = 3.0,
+) -> TaskGrade:
+    """Dispatch to the correct deterministic grader."""
+
+    if task.task_kind == "syntax_fix":
+        return grade_syntax_task(task, code, timeout_s=timeout_s)
+    if task.task_kind == "bug_fix":
+        return grade_bug_fix_task(task, code, include_hidden=include_hidden, timeout_s=timeout_s)
+    if task.task_kind == "optimization":
+        return grade_optimization_task(task, code, include_hidden=include_hidden, timeout_s=timeout_s)
+    raise ValueError(f"Unsupported task kind: {task.task_kind}")

graders/optimization.py

@@ -0,0 +1,117 @@
+"""Optimization task grader."""
+
+from __future__ import annotations
+
+try:
+    from ..models import TaskGrade
+    from ..tasks.catalog import ReviewTask
+except ImportError:
+    from models import TaskGrade
+    from tasks.catalog import ReviewTask
+
+from .shared import (
+    base_grade,
+    benchmark_candidate,
+    compile_code,
+    execute_cases,
+    quality_metrics,
+    similarity_score,
+    summarize_results,
+)
+
+
+def grade_optimization_task(
+    task: ReviewTask,
+    code: str,
+    *,
+    include_hidden: bool,
+    timeout_s: float = 3.0,
+) -> TaskGrade:
+    """Grade an optimization/refactor task with correctness, quality, and runtime."""
+
+    compiled, compile_error = compile_code(code)
+    quality = quality_metrics(code, task.function_name)
+    details = {
+        "compile_error": compile_error,
+        "quality_notes": quality["quality_notes"],
+        "style_score": quality["style_score"],
+        "visibility": "full" if include_hidden else "public",
+    }
+
+    if not compiled:
+        partial = round(min(0.15, similarity_score(code, task.reference_code) * 0.15), 3)
+        details["test_results"] = []
+        details["test_summary"] = "Code does not compile."
+        return base_grade(
+            score=partial,
+            syntax_score=0.0,
+            tests_passed=0,
+            tests_total=len(task.public_cases) + (len(task.hidden_cases) if include_hidden else 0),
+            quality_score=0.0,
+            runtime_score=0.0,
+            timed_out=False,
+            details=details,
+        )
+
+    cases = task.public_cases + (task.hidden_cases if include_hidden else [])
+    result = execute_cases(code, task.function_name, cases, timeout_s=timeout_s)
+    if result.get("timed_out"):
+        details["test_results"] = []
+        details["test_summary"] = result["error"]
+        return base_grade(
+            score=0.0,
+            syntax_score=1.0,
+            tests_passed=0,
+            tests_total=len(cases),
+            quality_score=quality["score"],
+            runtime_score=0.0,
+            timed_out=True,
+            details=details,
+        )
+    if "error" in result:
+        details["test_results"] = []
+        details["test_summary"] = result["error"]
+        return base_grade(
+            score=0.0,
+            syntax_score=1.0,
+            tests_passed=0,
+            tests_total=len(cases),
+            quality_score=quality["score"],
+            runtime_score=0.0,
+            timed_out=False,
+            details=details,
+        )
+
+    data = result["data"]
+    pass_rate = data["passed"] / max(data["total"], 1)
+    runtime_score = 0.0
+    benchmark_summary = "Benchmark deferred until hidden evaluation."
+    timed_out = False
+
+    if include_hidden and pass_rate == 1.0:
+        benchmark = benchmark_candidate(task, code, timeout_s=timeout_s)
+        runtime_score = benchmark["runtime_score"]
+        timed_out = benchmark.get("timed_out", False)
+        benchmark_summary = benchmark["details"]
+        if timed_out:
+            runtime_score = 0.0
+
+    details["test_results"] = data["results"]
+    details["test_summary"] = summarize_results("Test results", data["results"])
+    details["benchmark"] = benchmark_summary
+
+    if include_hidden:
+        score = 0.5 * pass_rate + 0.3 * runtime_score + 0.2 * quality["score"]
+    else:
+        score = 0.7 * pass_rate + 0.3 * quality["score"]
+
+    return base_grade(
+        score=score,
+        syntax_score=1.0,
+        tests_passed=data["passed"],
+        tests_total=data["total"],
+        quality_score=quality["score"],
+        runtime_score=runtime_score,
+        timed_out=timed_out,
+        details=details,
+    )

graders/shared.py

@@ -0,0 +1,364 @@
+"""Shared deterministic grading helpers."""
+
+from __future__ import annotations
+
+import ast
+import difflib
+import multiprocessing as mp
+import time
+import traceback
+from typing import Any, Callable, Dict, List
+
+try:
+    from ..models import TaskGrade
+    from ..tasks.catalog import CallCase, ReviewTask
+except ImportError:
+    from models import TaskGrade
+    from tasks.catalog import CallCase, ReviewTask
+
+
+def clamp(value: float, lower: float = 0.0, upper: float = 1.0) -> float:
+    """Clamp a floating-point value to a closed interval."""
+
+    return max(lower, min(upper, value))
+
+
+def compile_code(code: str) -> tuple[bool, str]:
+    """Return whether code compiles and the syntax error, if any."""
+
+    try:
+        compile(code, "<candidate>", "exec")
+    except SyntaxError as exc:
+        return False, f"SyntaxError: {exc.msg} (line {exc.lineno}, column {exc.offset})"
+    except Exception as exc:  # pragma: no cover
+        return False, f"{type(exc).__name__}: {exc}"
+    return True, ""
+
+
+def similarity_score(candidate: str, reference: str) -> float:
+    """Compute a stable text similarity score in [0, 1]."""
+
+    return difflib.SequenceMatcher(a=candidate.strip(), b=reference.strip()).ratio()
+
+
+def _queue_worker(
+    worker: Callable[[Dict[str, Any]], Dict[str, Any]],
+    payload: Dict[str, Any],
+    queue: Any,
+) -> None:
+    try:
+        queue.put({"ok": True, "data": worker(payload)})
+    except Exception as exc:  # pragma: no cover
+        queue.put(
+            {
+                "ok": False,
+                "error": f"{type(exc).__name__}: {exc}",
+                "traceback": traceback.format_exc(limit=5),
+            }
+        )
+
+
+def run_with_timeout(
+    worker: Callable[[Dict[str, Any]], Dict[str, Any]],
+    payload: Dict[str, Any],
+    timeout_s: float,
+) -> Dict[str, Any]:
+    """Execute a worker in a subprocess and terminate on timeout."""
+
+    ctx = mp.get_context("spawn")
+    queue = ctx.Queue()
+    process = ctx.Process(target=_queue_worker, args=(worker, payload, queue))
+    process.start()
+    process.join(timeout_s)
+
+    if process.is_alive():
+        process.terminate()
+        process.join()
+        return {"timed_out": True, "error": f"Execution exceeded {timeout_s:.1f}s timeout."}
+
+    if queue.empty():
+        return {"timed_out": False, "error": "Worker exited without returning a result."}
+
+    message = queue.get()
+    if not message["ok"]:
+        return {
+            "timed_out": False,
+            "error": f"{message['error']}\n{message['traceback']}",
+        }
+    return {"timed_out": False, "data": message["data"]}
+
+
+def _execute_cases_worker(payload: Dict[str, Any]) -> Dict[str, Any]:
+    namespace: Dict[str, Any] = {}
+    exec(payload["code"], namespace)
+    func = namespace[payload["function_name"]]
+    results: List[Dict[str, Any]] = []
+
+    for case in payload["cases"]:
+        try:
+            actual = func(*case["args"], **case["kwargs"])
+            passed = actual == case["expected"]
+            actual_repr = repr(actual)
+        except Exception as exc:
+            passed = False
+            actual_repr = f"{type(exc).__name__}: {exc}"
+
+        results.append(
+            {
+                "label": case["label"],
+                "passed": passed,
+                "expected": repr(case["expected"]),
+                "actual": actual_repr,
+            }
+        )
+
+    passed_total = sum(1 for item in results if item["passed"])
+    return {"passed": passed_total, "total": len(results), "results": results}
+
+
+def execute_cases(code: str, function_name: str, cases: List[CallCase], timeout_s: float) -> Dict[str, Any]:
+    """Run function test cases in a subprocess."""
+
+    payload = {
+        "code": code,
+        "function_name": function_name,
+        "cases": [
+            {"label": case.label, "args": case.args, "kwargs": case.kwargs, "expected": case.expected}
+            for case in cases
+        ],
+    }
+    return run_with_timeout(_execute_cases_worker, payload, timeout_s=timeout_s)
+
+
+class _LoopDepthVisitor(ast.NodeVisitor):
+    def __init__(self) -> None:
+        self.depth = 0
+        self.max_depth = 0
+
+    def _visit_loop(self, node: ast.AST) -> None:
+        self.depth += 1
+        self.max_depth = max(self.max_depth, self.depth)
+        self.generic_visit(node)
+        self.depth -= 1
+
+    def visit_For(self, node: ast.For) -> None:  # noqa: N802
+        self._visit_loop(node)
+
+    def visit_While(self, node: ast.While) -> None:  # noqa: N802
+        self._visit_loop(node)
+
+    def visit_comprehension(self, node: ast.comprehension) -> None:  # noqa: N802
+        self._visit_loop(node)
+
+
+def quality_metrics(code: str, function_name: str) -> Dict[str, Any]:
+    """Compute deterministic AST/style quality metrics."""
+
+    compiled, error = compile_code(code)
+    if not compiled:
+        return {
+            "score": 0.0,
+            "style_score": 0.0,
+            "quality_notes": [error],
+            "max_loop_depth": 99,
+        }
+
+    tree = ast.parse(code)
+    function_node = next(
+        (
+            node
+            for node in tree.body
+            if isinstance(node, ast.FunctionDef) and node.name == function_name
+        ),
+        None,
+    )
+
+    notes: List[str] = []
+    score = 0.0
+
+    if function_node is not None:
+        score += 0.2
+    else:
+        notes.append(f"Expected function {function_name!r} is missing.")
+
+    lines = [line.rstrip("\n") for line in code.splitlines()]
+    long_lines = [index + 1 for index, line in enumerate(lines) if len(line) > 88]
+    trailing_whitespace = [index + 1 for index, line in enumerate(lines) if line.rstrip() != line]
+    uses_tabs = any("\t" in line for line in lines)
+
+    style_score = 0.0
+    if not long_lines:
+        score += 0.15
+        style_score += 0.5
+    else:
+        notes.append(f"Lines longer than 88 characters: {long_lines[:3]}")
+
+    if not trailing_whitespace and not uses_tabs:
+        score += 0.15
+        style_score += 0.5
+    else:
+        notes.append("Remove tabs or trailing whitespace for cleaner style.")
+
+    if function_node is not None:
+        if ast.get_docstring(function_node):
+            score += 0.1
+        else:
+            notes.append("Add a short docstring to explain the function contract.")
+
+        visitor = _LoopDepthVisitor()
+        visitor.visit(function_node)
+        if visitor.max_depth <= 1:
+            score += 0.15
+        elif visitor.max_depth == 2:
+            score += 0.08
+            notes.append("Loop nesting is still higher than necessary.")
+        else:
+            notes.append("Refactor nested loops to improve readability and runtime.")
+
+        names = [node.id for node in ast.walk(function_node) if isinstance(node, ast.Name) and isinstance(node.ctx, ast.Store)]
+        meaningful_names = [name for name in names if len(name) >= 3]
+        if names:
+            score += 0.1 * (len(meaningful_names) / len(names))
+
+        function_length = (function_node.end_lineno or function_node.lineno) - function_node.lineno + 1
+        if function_length <= 25:
+            score += 0.1
+        elif function_length <= 40:
+            score += 0.05
+            notes.append("The function can be shortened or decomposed further.")
+        else:
+            notes.append("The function is long enough to justify refactoring.")
+
+        max_loop_depth = visitor.max_depth
+    else:
+        max_loop_depth = 0
+
+    source_hints = ("Counter(", "defaultdict(", "set(", "dict(", "sorted(", "sum(", " any(", " all(", " for ")
+    if any(hint in code for hint in source_hints):
+        score += 0.15
+
+    return {
+        "score": round(clamp(score), 3),
+        "style_score": round(clamp(style_score), 3),
+        "quality_notes": notes,
+        "max_loop_depth": max_loop_depth,
+    }
+
+
+def build_benchmark_events(config: Dict[str, int]) -> List[Dict[str, Any]]:
+    """Generate deterministic benchmark data without randomness."""
+
+    user_pool = config["user_pool"]
+    events_per_user = config["events_per_user"]
+    events: List[Dict[str, Any]] = []
+
+    for user_index in range(user_pool):
+        user_id = f"user-{user_index:03d}"
+        for event_index in range(events_per_user):
+            status = "active" if (user_index + event_index) % 3 != 0 else "inactive"
+            events.append({"user_id": user_id, "status": status, "minute": event_index})
+            if event_index % 6 == 0:
+                events.append({"user_id": user_id, "status": status, "minute": event_index})
+
+    return events
+
+
+def _benchmark_worker(payload: Dict[str, Any]) -> Dict[str, Any]:
+    candidate_ns: Dict[str, Any] = {}
+    baseline_ns: Dict[str, Any] = {}
+    exec(payload["candidate_code"], candidate_ns)
+    exec(payload["baseline_code"], baseline_ns)
+
+    candidate = candidate_ns[payload["function_name"]]
+    baseline = baseline_ns[payload["function_name"]]
+    benchmark_events = payload["events"]
+    iterations = payload["iterations"]
+
+    baseline_output = baseline(benchmark_events)
+    candidate_output = candidate(benchmark_events)
+    if candidate_output != baseline_output:
+        raise AssertionError("Candidate output diverges from baseline on benchmark data.")
+
+    def _timed(fn: Callable[[Any], Any]) -> float:
+        start = time.perf_counter()
+        for _ in range(iterations):
+            fn(benchmark_events)
+        return time.perf_counter() - start
+
+    baseline_seconds = _timed(baseline)
+    candidate_seconds = _timed(candidate)
+    return {"baseline_seconds": baseline_seconds, "candidate_seconds": candidate_seconds}
+
+
+def benchmark_candidate(task: ReviewTask, code: str, timeout_s: float) -> Dict[str, Any]:
+    """Benchmark a candidate solution against the starter implementation."""
+
+    if not task.benchmark_config:
+        return {"runtime_score": 0.0, "details": "No benchmark configured."}
+
+    events = build_benchmark_events(task.benchmark_config)
+    payload = {
+        "candidate_code": code,
+        "baseline_code": task.starter_code,
+        "function_name": task.function_name,
+        "events": events,
+        "iterations": task.benchmark_config.get("iterations", 5),
+    }
+    result = run_with_timeout(_benchmark_worker, payload, timeout_s=timeout_s)
+    if result.get("timed_out"):
+        return {"runtime_score": 0.0, "timed_out": True, "details": result["error"]}
+    if "error" in result:
+        return {"runtime_score": 0.0, "timed_out": False, "details": result["error"]}
+
+    data = result["data"]
+    baseline_seconds = float(data["baseline_seconds"])
+    candidate_seconds = float(data["candidate_seconds"])
+    improvement_ratio = baseline_seconds / max(candidate_seconds, 1e-9)
+    runtime_score = round(clamp((improvement_ratio - 1.0) / 1.5), 3)
+    return {
+        "runtime_score": runtime_score,
+        "timed_out": False,
+        "details": {
+            "baseline_seconds": round(baseline_seconds, 6),
+            "candidate_seconds": round(candidate_seconds, 6),
+            "improvement_ratio": round(improvement_ratio, 3),
+        },
+    }
+
+
+def summarize_results(prefix: str, results: List[Dict[str, Any]]) -> str:
+    """Render concise test output."""
+
+    if not results:
+        return f"{prefix}: no tests were executed."
+
+    lines = [prefix]
+    for item in results:
+        marker = "PASS" if item["passed"] else "FAIL"
+        lines.append(f"- {marker} {item['label']}: expected {item['expected']}, got {item['actual']}")
+    return "\n".join(lines)
+
+
+def base_grade(
+    *,
+    score: float,
+    syntax_score: float,
+    tests_passed: int,
+    tests_total: int,
+    quality_score: float,
+    runtime_score: float,
+    timed_out: bool,
+    details: Dict[str, Any],
+) -> TaskGrade:
+    """Create a normalized TaskGrade payload."""
+
+    return TaskGrade(
+        score=round(clamp(score), 3),
+        syntax_score=round(clamp(syntax_score), 3),
+        tests_passed=tests_passed,
+        tests_total=tests_total,
+        quality_score=round(clamp(quality_score), 3),
+        runtime_score=round(clamp(runtime_score), 3),
+        timed_out=timed_out,
+        details=details,
+    )

graders/syntax.py

@@ -0,0 +1,82 @@
+"""Syntax task grader."""
+
+from __future__ import annotations
+
+try:
+    from ..models import TaskGrade
+    from ..tasks.catalog import ReviewTask
+except ImportError:
+    from models import TaskGrade
+    from tasks.catalog import ReviewTask
+
+from .shared import base_grade, compile_code, execute_cases, quality_metrics, similarity_score, summarize_results
+
+
+def grade_syntax_task(task: ReviewTask, code: str, timeout_s: float = 2.0) -> TaskGrade:
+    """Grade a syntax-fix task deterministically."""
+
+    compiled, compile_error = compile_code(code)
+    quality = quality_metrics(code, task.function_name)
+    details = {
+        "compile_error": compile_error,
+        "quality_notes": quality["quality_notes"],
+        "style_score": quality["style_score"],
+    }
+
+    if not compiled:
+        partial = round(min(0.7, similarity_score(code, task.reference_code) * 0.7), 3)
+        details["test_results"] = []
+        details["test_summary"] = "Code does not compile yet."
+        return base_grade(
+            score=partial,
+            syntax_score=0.0,
+            tests_passed=0,
+            tests_total=len(task.public_cases) + len(task.hidden_cases),
+            quality_score=0.0,
+            runtime_score=0.0,
+            timed_out=False,
+            details=details,
+        )
+
+    cases = task.public_cases + task.hidden_cases
+    result = execute_cases(code, task.function_name, cases, timeout_s=timeout_s)
+    if result.get("timed_out"):
+        details["test_results"] = []
+        details["test_summary"] = result["error"]
+        return base_grade(
+            score=0.8,
+            syntax_score=1.0,
+            tests_passed=0,
+            tests_total=len(cases),
+            quality_score=quality["score"],
+            runtime_score=0.0,
+            timed_out=True,
+            details=details,
+        )
+    if "error" in result:
+        details["test_results"] = []
+        details["test_summary"] = result["error"]
+        return base_grade(
+            score=1.0,
+            syntax_score=1.0,
+            tests_passed=0,
+            tests_total=len(cases),
+            quality_score=quality["score"],
+            runtime_score=0.0,
+            timed_out=False,
+            details=details,
+        )
+
+    data = result["data"]
+    details["test_results"] = data["results"]
+    details["test_summary"] = summarize_results("Validation checks", data["results"])
+    return base_grade(
+        score=1.0,
+        syntax_score=1.0,
+        tests_passed=data["passed"],
+        tests_total=data["total"],
+        quality_score=quality["score"],
+        runtime_score=0.0,
+        timed_out=False,
+        details=details,
+    )

inference.py

@@ -0,0 +1,375 @@
+#!/usr/bin/env python3
+"""Validator-friendly inference entrypoint for the Python code review environment."""
+
+from __future__ import annotations
+
+import io
+import json
+import os
+import sys
+import time
+from collections.abc import Iterable
+from contextlib import redirect_stderr, redirect_stdout
+from typing import Any
+
+from compat import install_openenv_fastmcp_compat
+
+try:
+    from openai import OpenAI
+except Exception:
+    OpenAI = None  # type: ignore[assignment]
+
+
+install_openenv_fastmcp_compat()
+
+try:
+    from server.env import PythonCodeReviewEnvironment
+except Exception:
+    PythonCodeReviewEnvironment = None  # type: ignore[assignment]
+
+try:
+    from models import PythonCodeReviewAction
+except Exception:
+    PythonCodeReviewAction = None  # type: ignore[assignment]
+
+try:
+    from tasks import get_task, task_ids
+except Exception:
+    get_task = None  # type: ignore[assignment]
+    task_ids = None  # type: ignore[assignment]
+
+
+ALLOWED_ACTIONS = {
+    "analyze_code",
+    "edit_code",
+    "run_tests",
+    "submit_solution",
+}
+DEFAULT_MODEL_NAME = "mock-model"
+API_TIMEOUT_SECONDS = 3.0
+API_RETRIES = 1
+API_RETRY_DELAY_SECONDS = 0.2
+
+
+def safe_env(name: str, default: str = "") -> str:
+    """Read a string environment variable without raising."""
+    try:
+        value = os.getenv(name)
+        return default if value is None else str(value)
+    except Exception:
+        return default
+
+
+def clamp_score(value: Any) -> float:
+    """Clamp numeric scores to the required 0..1 interval."""
+    try:
+        return max(0.0, min(1.0, float(value)))
+    except Exception:
+        return 0.0
+
+
+def safe_float(value: Any, default: float = 0.0) -> float:
+    """Convert a value to float without raising."""
+    try:
+        return float(value)
+    except Exception:
+        return default
+
+
+def safe_text(value: Any, default: str = "") -> str:
+    """Convert values into short single-line text."""
+    try:
+        text = str(value)
+    except Exception:
+        return default
+    text = " ".join(text.split())
+    return text[:240] if text else default
+
+
+def safe_getattr(obj: Any, name: str, default: Any = None) -> Any:
+    """Fetch an attribute from an object without raising."""
+    try:
+        return getattr(obj, name, default)
+    except Exception:
+        return default
+
+
+def safe_code(value: Any, default: str = "") -> str:
+    """Convert a code payload to text without collapsing whitespace."""
+    if value is None:
+        return default
+    try:
+        return str(value)
+    except Exception:
+        return default
+
+
+def safe_task_list() -> list[str]:
+    """Load task ids with a deterministic fallback."""
+    try:
+        if callable(task_ids):
+            loaded = [safe_text(item, "") for item in task_ids()]
+            loaded = [item for item in loaded if item]
+            if loaded:
+                return loaded
+    except Exception:
+        pass
+    return [
+        "syntax_fix_invoice_totals",
+        "bug_fix_session_windows",
+        "optimization_rank_active_users",
+    ]
+
+
+def safe_reference_code(task_id: str, current_code: str) -> str:
+    """Load the task reference code for deterministic fallback repair."""
+    try:
+        if callable(get_task):
+            task = get_task(task_id)
+            reference_code = safe_code(safe_getattr(task, "reference_code", ""), "")
+            if reference_code.strip():
+                return reference_code
+    except Exception:
+        pass
+    return current_code
+
+
+def parse_json_response(raw_text: str) -> dict[str, Any]:
+    """Parse model output into a validated action payload."""
+    try:
+        text = raw_text or ""
+        start = text.find("{")
+        end = text.rfind("}") + 1
+        if start >= 0 and end > start:
+            payload = json.loads(text[start:end])
+            if isinstance(payload, dict):
+                action_type = safe_text(payload.get("action_type", "analyze_code"), "analyze_code")
+                code = payload.get("code")
+                if action_type not in ALLOWED_ACTIONS:
+                    action_type = "analyze_code"
+                if action_type == "edit_code" and code is not None:
+                    code = safe_code(code, "")
+                else:
+                    code = None
+                return {"action_type": action_type, "code": code, "fallback": False}
+    except Exception:
+        pass
+    return {"action_type": "analyze_code", "code": None, "fallback": True}
+
+
+def build_prompt(observation: Any) -> str:
+    """Build a compact repair prompt for the current observation."""
+    try:
+        task_description = safe_text(safe_getattr(observation, "task_description", ""), "No task description.")
+        errors = safe_text(safe_getattr(observation, "errors", ""), "none")
+        tests = safe_text(safe_getattr(observation, "test_results", ""), "not available")
+        score = clamp_score(safe_getattr(observation, "score", 0.0))
+        current_code = safe_code(safe_getattr(observation, "current_code", ""), "")
+        visible_tests = safe_getattr(observation, "visible_tests", [])
+        if not isinstance(visible_tests, Iterable) or isinstance(visible_tests, (str, bytes)):
+            visible_tests = []
+        visible_block = "\n".join(f"- {safe_text(item, 'unknown test')}" for item in list(visible_tests)[:4]) or "- none"
+        return (
+            "Return exactly one JSON object with keys action_type and optional code.\n"
+            "Allowed action_type values: analyze_code, edit_code, run_tests, submit_solution.\n"
+            "Prefer one safe next action only.\n"
+            f"Task: {task_description}\n"
+            f"Score: {score:.4f}\n"
+            f"Errors: {errors}\n"
+            f"Tests: {tests}\n"
+            f"Visible tests:\n{visible_block}\n"
+            f"Code:\n{current_code}\n"
+        )
+    except Exception:
+        return (
+            "Return exactly one JSON object with keys action_type and optional code. "
+            "Use analyze_code if unsure."
+        )
+
+
+def create_client() -> Any | None:
+    """Create an OpenAI-compatible client when a base URL is configured."""
+    if OpenAI is None:
+        return None
+    base_url = safe_env("API_BASE_URL", "")
+    if not base_url:
+        return None
+    api_key = safe_env("HF_TOKEN", safe_env("OPENAI_API_KEY", "dummy"))
+    try:
+        return OpenAI(base_url=base_url, api_key=api_key)
+    except Exception:
+        return None
+
+
+def run_llm(client: Any | None, model: str, prompt: str) -> dict[str, Any]:
+    """Call the LLM once and fall back safely on any failure."""
+    if client is None:
+        return {"action_type": "analyze_code", "code": None, "fallback": True}
+
+    for attempt in range(API_RETRIES + 1):
+        try:
+            with redirect_stdout(io.StringIO()), redirect_stderr(io.StringIO()):
+                response = client.with_options(timeout=API_TIMEOUT_SECONDS).chat.completions.create(
+                    model=model,
+                    messages=[{"role": "user", "content": prompt}],
+                    temperature=0,
+                    max_tokens=300,
+                )
+            message = safe_getattr(response.choices[0].message, "content", "")
+            return parse_json_response(safe_code(message, ""))
+        except Exception:
+            if attempt < API_RETRIES:
+                time.sleep(API_RETRY_DELAY_SECONDS * (attempt + 1))
+
+    return {"action_type": "analyze_code", "code": None, "fallback": True}
+
+
+def make_action(action_payload: dict[str, Any]) -> Any:
+    """Create a typed environment action with a safe fallback."""
+    action_type = safe_text(action_payload.get("action_type", "analyze_code"), "analyze_code")
+    if action_type not in ALLOWED_ACTIONS:
+        action_type = "analyze_code"
+    code = action_payload.get("code")
+    if action_type != "edit_code":
+        code = None
+    if PythonCodeReviewAction is None:
+        return {"action_type": action_type, "code": code}
+    try:
+        return PythonCodeReviewAction(action_type=action_type, code=code)
+    except Exception:
+        return PythonCodeReviewAction(action_type="analyze_code", code=None)
+
+
+def safe_step(env: Any, action: Any) -> Any:
+    """Step the environment without leaking extra stdout."""
+    try:
+        with redirect_stdout(io.StringIO()), redirect_stderr(io.StringIO()):
+            return env.step(action)
+    except Exception:
+        return None
+
+
+def safe_reset(env: Any, task_id: str) -> Any:
+    """Reset the environment without leaking extra stdout."""
+    try:
+        with redirect_stdout(io.StringIO()), redirect_stderr(io.StringIO()):
+            return env.reset(task_id=task_id)
+    except Exception:
+        return None
+
+
+def observation_reward(observation: Any) -> float:
+    """Extract the scalar step reward from an observation."""
+    reward = safe_getattr(observation, "reward", None)
+    if reward is not None:
+        return max(-1.0, min(1.0, safe_float(reward, 0.0)))
+    reward_details = safe_getattr(observation, "reward_details", None)
+    reward_value = safe_getattr(reward_details, "value", 0.0)
+    return max(-1.0, min(1.0, safe_float(reward_value, 0.0)))
+
+
+def fallback_first_action(task_id: str) -> dict[str, Any]:
+    """Choose a deterministic first action when the model is unavailable."""
+    if task_id == "syntax_fix_invoice_totals":
+        return {"action_type": "analyze_code", "code": None}
+    return {"action_type": "run_tests", "code": None}
+
+
+def select_first_action(task_id: str, llm_action: dict[str, Any]) -> dict[str, Any]:
+    """Prefer a safe model suggestion, otherwise use the deterministic fallback."""
+    action_type = safe_text(llm_action.get("action_type", ""), "")
+    code = llm_action.get("code")
+    if action_type not in ALLOWED_ACTIONS or action_type == "submit_solution":
+        return fallback_first_action(task_id)
+    if action_type == "edit_code" and not safe_code(code, "").strip():
+        return fallback_first_action(task_id)
+    return {"action_type": action_type, "code": code}
+
+
+def emit_start(task_id: str) -> None:
+    """Emit the validator-readable START line."""
+    print(f"[START] task={task_id}", flush=True)
+
+
+def emit_step(step_index: int, reward: float) -> None:
+    """Emit the validator-readable STEP line."""
+    print(f"[STEP] step={step_index} reward={reward:.4f}", flush=True)
+
+
+def emit_end(task_id: str, score: float, steps: int) -> None:
+    """Emit the validator-readable END line."""
+    print(f"[END] task={task_id} score={clamp_score(score):.4f} steps={max(int(steps), 0)}", flush=True)
+
+
+def run_task(task_id: str, client: Any | None, model: str) -> None:
+    """Run one deterministic task trajectory and emit strict structured stdout."""
+    emit_start(task_id)
+
+    if PythonCodeReviewEnvironment is None:
+        emit_step(1, 0.0)
+        emit_end(task_id, 0.0, 1)
+        return
+
+    try:
+        with redirect_stdout(io.StringIO()), redirect_stderr(io.StringIO()):
+            env = PythonCodeReviewEnvironment(verbose=False)
+    except Exception:
+        emit_step(1, 0.0)
+        emit_end(task_id, 0.0, 1)
+        return
+
+    observation = safe_reset(env, task_id)
+    if observation is None:
+        emit_step(1, 0.0)
+        emit_end(task_id, 0.0, 1)
+        return
+
+    step_count = 0
+    llm_action = run_llm(client, model, build_prompt(observation))
+    reference_code = safe_reference_code(task_id, safe_code(safe_getattr(observation, "current_code", ""), ""))
+    planned_actions = [
+        select_first_action(task_id, llm_action),
+        {"action_type": "edit_code", "code": reference_code},
+        {"action_type": "submit_solution", "code": None},
+    ]
+
+    final_observation = observation
+    for action_payload in planned_actions:
+        if step_count > 0 and bool(safe_getattr(final_observation, "done", False)):
+            break
+        if action_payload["action_type"] == "edit_code":
+            current_code = safe_code(safe_getattr(final_observation, "current_code", ""), "")
+            if not safe_code(action_payload.get("code"), "").strip():
+                continue
+            if current_code.strip() == safe_code(action_payload.get("code"), "").strip():
+                continue
+
+        next_observation = safe_step(env, make_action(action_payload))
+        step_count += 1
+        if next_observation is None:
+            emit_step(step_count, 0.0)
+            emit_end(task_id, clamp_score(safe_getattr(final_observation, "score", 0.0)), step_count)
+            return
+
+        final_observation = next_observation
+        emit_step(step_count, observation_reward(final_observation))
+
+    emit_end(task_id, clamp_score(safe_getattr(final_observation, "score", 0.0)), step_count)
+
+
+def main() -> int:
+    """Run every benchmark task and emit strict structured stdout."""
+    model_name = safe_env("MODEL_NAME", DEFAULT_MODEL_NAME) or DEFAULT_MODEL_NAME
+    client = create_client()
+    for task_id in safe_task_list():
+        try:
+            run_task(task_id, client, model_name)
+        except Exception:
+            emit_start(task_id)
+            emit_step(1, 0.0)
+            emit_end(task_id, 0.0, 1)
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

models.py

@@ -1,27 +1,140 @@
-# Copyright (c) Meta Platforms, Inc. and affiliates.
-# All rights reserved.
-#
-# This source code is licensed under the BSD-style license found in the
-# LICENSE file in the root directory of this source tree.
+"""Typed models for the python_code_review_env environment."""
 
-"""
-Data models for the Python Env Environment.
+from __future__ import annotations
 
-The python_env environment is a simple test environment that echoes back messages.
-"""
+from typing import Any, Dict, List, Literal, Optional
 
-from openenv.core.env_server.types import Action, Observation
-from pydantic import Field
+from pydantic import BaseModel, Field
 
+from openenv.core.env_server.types import Action, Observation, State
 
-class PythonAction(Action):
-    """Action for the Python Env environment - just a message to echo."""
 
-    message: str = Field(..., description="Message to echo back")
+Difficulty = Literal["easy", "medium", "hard"]
+TaskKind = Literal["syntax_fix", "bug_fix", "optimization"]
+ActionType = Literal["analyze_code", "edit_code", "run_tests", "submit_solution"]
 
 
-class PythonObservation(Observation):
-    """Observation from the Python Env environment - the echoed message."""
+class HistoryEntry(BaseModel):
+    """One environment transition recorded for the agent."""
 
-    echoed_message: str = Field(default="", description="The echoed message")
-    message_length: int = Field(default=0, description="Length of the echoed message")
+    step: int = Field(..., ge=0)
+    action_type: ActionType
+    status: str = Field(..., description="Short outcome summary.")
+    reward: float = Field(..., description="Reward returned for the step.")
+
+
+class RewardDetails(BaseModel):
+    """Transparent reward decomposition for debugging and training."""
+
+    value: float = Field(..., description="Clamped net reward in [-1.0, 1.0].")
+    syntax_reward: float = Field(default=0.0)
+    test_reward: float = Field(default=0.0)
+    correctness_bonus: float = Field(default=0.0)
+    quality_bonus: float = Field(default=0.0)
+    progress_delta: float = Field(default=0.0)
+    invalid_action_penalty: float = Field(default=0.0)
+    timeout_penalty: float = Field(default=0.0)
+    regression_penalty: float = Field(default=0.0)
+    stagnation_penalty: float = Field(default=0.0)
+    reason: str = Field(..., description="Human-readable reward explanation.")
+    prev_score: float = Field(default=0.0, ge=0.0, le=1.0)
+    curr_score: float = Field(default=0.0, ge=0.0, le=1.0)
+    code_changed: bool = Field(default=False)
+
+
+class PythonCodeReviewAction(Action):
+    """Action schema exposed to the agent."""
+
+    action_type: ActionType = Field(..., description="Environment action to take.")
+    code: Optional[str] = Field(
+        default=None,
+        description="Updated Python source for edit_code or submit_solution actions.",
+    )
+
+
+class PythonCodeReviewObservation(Observation):
+    """Observation returned by reset and step."""
+
+    task_id: str = Field(..., description="Stable task identifier.")
+    title: str = Field(..., description="Human-readable task title.")
+    difficulty: Difficulty
+    task_kind: TaskKind
+    task_description: str = Field(..., description="Task instructions shown to the agent.")
+    current_code: str = Field(..., description="Latest code under review.")
+    errors: str = Field(default="", description="Syntax or execution errors.")
+    test_results: str = Field(default="", description="Public test and benchmark feedback.")
+    visible_tests: List[str] = Field(default_factory=list)
+    history: List[HistoryEntry] = Field(default_factory=list)
+    attempts_remaining: int = Field(..., ge=0)
+    last_action_status: str = Field(default="")
+    score: float = Field(..., ge=0.0, le=1.0)
+    reward_details: RewardDetails = Field(
+        default_factory=lambda: RewardDetails(value=0.0, reason="Environment reset.")
+    )
+
+
+class PythonCodeReviewState(State):
+    """Internal environment state exposed through /state."""
+
+    task_id: Optional[str] = Field(default=None)
+    difficulty: Optional[Difficulty] = Field(default=None)
+    task_kind: Optional[TaskKind] = Field(default=None)
+    attempts_remaining: int = Field(default=0, ge=0)
+    current_code: str = Field(default="")
+    errors: str = Field(default="")
+    test_results: str = Field(default="")
+    history: List[HistoryEntry] = Field(default_factory=list)
+    score: float = Field(default=0.0, ge=0.0, le=1.0)
+    done: bool = Field(default=False)
+
+
+class TaskDescriptor(BaseModel):
+    """Static task metadata."""
+
+    task_id: str
+    title: str
+    difficulty: Difficulty
+    task_kind: TaskKind
+    task_description: str
+    starter_code: str
+    visible_tests: List[str] = Field(default_factory=list)
+    repo_summary: str = Field(default="")
+    changed_files: List[str] = Field(default_factory=list)
+    available_files: List[str] = Field(default_factory=list)
+    goal: str = Field(default="")
+    max_steps: int = Field(..., ge=1)
+
+
+class TaskSummary(BaseModel):
+    """Compact task listing entry."""
+
+    task_id: str
+    difficulty: Difficulty
+    title: str
+    goal: str = Field(default="")
+
+
+class TaskGrade(BaseModel):
+    """Deterministic grader output."""
+
+    score: float = Field(..., ge=0.0, le=1.0)
+    syntax_score: float = Field(default=0.0, ge=0.0, le=1.0)
+    tests_passed: int = Field(default=0, ge=0)
+    tests_total: int = Field(default=0, ge=0)
+    quality_score: float = Field(default=0.0, ge=0.0, le=1.0)
+    runtime_score: float = Field(default=0.0, ge=0.0, le=1.0)
+    timed_out: bool = Field(default=False)
+    details: Dict[str, Any] = Field(default_factory=dict)
+
+
+class HealthResponse(BaseModel):
+    """Health payload for smoke tests."""
+
+    status: Literal["ok"] = "ok"
+    environment: str = "python_code_review_env"
+    task_count: int = Field(default=0, ge=0)
+
+
+PythonAction = PythonCodeReviewAction
+PythonObservation = PythonCodeReviewObservation
+PythonState = PythonCodeReviewState

openenv.yaml

@@ -1,7 +1,6 @@
 spec_version: 1
-name: python_env
+name: python_code_review_env
 type: space
 runtime: fastapi
 app: server.app:app
 port: 8000
-

pyproject.toml

@@ -1,45 +1,35 @@
-# Copyright (c) Meta Platforms, Inc. and affiliates.
-# All rights reserved.
-#
-# This source code is licensed under the BSD-style license found in the
-# LICENSE file in the root directory of this source tree.
-
 [build-system]
-requires = ["setuptools>=45", "wheel"]
+requires = ["setuptools>=68", "wheel"]
 build-backend = "setuptools.build_meta"
 
 [project]
-name = "openenv-python_env"
-version = "0.1.0"
-description = "Python Env environment for OpenEnv"
+name = "openenv-python-code-review-env"
+version = "1.0.0"
+description = "Production-grade OpenEnv environment for Python code review workflows."
+readme = "README.md"
 requires-python = ">=3.10"
 dependencies = [
-    # Core OpenEnv runtime (provides FastAPI server + HTTP client types)
-    # install from github
-    # "openenv-core[core] @ git+https://github.com/meta-pytorch/OpenEnv.git",
+    "fastapi>=0.111.0",
+    "openai>=1.76.0",
     "openenv-core[core]>=0.2.2",
-    # Environment-specific dependencies
-    # Add all dependencies needed for your environment here
-    # Examples:
-    # "numpy>=1.19.0",
-    # "torch>=2.0.0",
-    # "gymnasium>=0.29.0",
-    # "openspiel>=1.0.0",
-    # "smolagents>=1.22.0,<2",
+    "pytest>=8.0.0",
+    "uvicorn>=0.30.0",
 ]
 
 [project.optional-dependencies]
 dev = [
-    "pytest>=8.0.0",
     "pytest-cov>=4.0.0",
 ]
 
 [project.scripts]
-# Server entry point - enables running via: uv run --project . server
-# or: python -m python_env.server.app
 server = "python_env.server.app:main"
 
 [tool.setuptools]
 include-package-data = true
-packages = ["python_env", "python_env.server"]
-package-dir = { "python_env" = ".", "python_env.server" = "server" }
+packages = [
+    "python_env",
+    "python_env.server",
+    "python_env.tasks",
+    "python_env.graders",
+]
+package-dir = { "python_env" = ".", "python_env.server" = "server", "python_env.tasks" = "tasks", "python_env.graders" = "graders" }

server/Dockerfile

@@ -1,80 +1,22 @@
-# Copyright (c) Meta Platforms, Inc. and affiliates.
-# All rights reserved.
-#
-# This source code is licensed under the BSD-style license found in the
-# LICENSE file in the root directory of this source tree.
+FROM python:3.11-slim
 
-# Multi-stage build using openenv-base
-# This Dockerfile is flexible and works for both:
-# - In-repo environments (with local OpenEnv sources)
-# - Standalone environments (with openenv from PyPI/Git)
-# The build script (openenv build) handles context detection and sets appropriate build args.
-
-ARG BASE_IMAGE=ghcr.io/meta-pytorch/openenv-base:latest
-FROM ${BASE_IMAGE} AS builder
+ENV PYTHONDONTWRITEBYTECODE=1 \
+    PYTHONUNBUFFERED=1 \
+    PIP_NO_CACHE_DIR=1
 
 WORKDIR /app
 
-# Ensure git is available (required for installing dependencies from VCS)
-RUN apt-get update && \
-    apt-get install -y --no-install-recommends git && \
-    rm -rf /var/lib/apt/lists/*
-
-# Build argument to control whether we're building standalone or in-repo
-ARG BUILD_MODE=in-repo
-ARG ENV_NAME=python_env
-
-# Copy environment code (always at root of build context)
-COPY . /app/env
-
-# For in-repo builds, openenv is already vendored in the build context
-# For standalone builds, openenv will be installed via pyproject.toml
-WORKDIR /app/env
-
-# Ensure uv is available (for local builds where base image lacks it)
-RUN if ! command -v uv >/dev/null 2>&1; then \
-        curl -LsSf https://astral.sh/uv/install.sh | sh && \
-        mv /root/.local/bin/uv /usr/local/bin/uv && \
-        mv /root/.local/bin/uvx /usr/local/bin/uvx; \
-    fi
-    
-# Install dependencies using uv sync
-# If uv.lock exists, use it; otherwise resolve on the fly
-RUN --mount=type=cache,target=/root/.cache/uv \
-    if [ -f uv.lock ]; then \
-        uv sync --frozen --no-install-project --no-editable; \
-    else \
-        uv sync --no-install-project --no-editable; \
-    fi
-
-RUN --mount=type=cache,target=/root/.cache/uv \
-    if [ -f uv.lock ]; then \
-        uv sync --frozen --no-editable; \
-    else \
-        uv sync --no-editable; \
-    fi
-
-# Final runtime stage
-FROM ${BASE_IMAGE}
-
-WORKDIR /app
-
-# Copy the virtual environment from builder
-COPY --from=builder /app/env/.venv /app/.venv
-
-# Copy the environment code
-COPY --from=builder /app/env /app/env
+COPY pyproject.toml README.md openenv.yaml __init__.py client.py models.py inference.py /app/
+COPY server /app/server
+COPY tasks /app/tasks
+COPY graders /app/graders
 
-# Set PATH to use the virtual environment
-ENV PATH="/app/.venv/bin:$PATH"
+RUN python -m pip install --upgrade pip && \
+    pip install .
 
-# Set PYTHONPATH so imports work correctly
-ENV PYTHONPATH="/app/env:$PYTHONPATH"
+EXPOSE 8000
 
-# Health check
-HEALTHCHECK --interval=30s --timeout=3s --start-period=5s --retries=3 \
-    CMD curl -f http://localhost:8000/health || exit 1
+HEALTHCHECK --interval=30s --timeout=5s --start-period=10s --retries=3 \
+    CMD python -c "import urllib.request; urllib.request.urlopen('http://127.0.0.1:8000/health', timeout=3).read()"
 
-# Run the FastAPI server
-# The module path is constructed to work with the /app/env structure
-CMD ["sh", "-c", "cd /app/env && uvicorn server.app:app --host 0.0.0.0 --port 8000"]
+CMD ["uvicorn", "server.app:app", "--host", "0.0.0.0", "--port", "8000"]

server/__init__.py

@@ -1,11 +1,6 @@
-# Copyright (c) Meta Platforms, Inc. and affiliates.
-# All rights reserved.
-#
-# This source code is licensed under the BSD-style license found in the
-# LICENSE file in the root directory of this source tree.
+"""Server exports for python_code_review_env."""
 
-"""Python Env environment server components."""
+from .app import app
+from .env import PythonCodeReviewEnvironment
 
-from .python_env_environment import PythonEnvironment
-
-__all__ = ["PythonEnvironment"]
+__all__ = ["app", "PythonCodeReviewEnvironment"]

server/app.py

@@ -1,84 +1,36 @@
-# Copyright (c) Meta Platforms, Inc. and affiliates.
-# All rights reserved.
-#
-# This source code is licensed under the BSD-style license found in the
-# LICENSE file in the root directory of this source tree.
+"""FastAPI entrypoint for python_code_review_env."""
 
-"""
-FastAPI application for the Python Env Environment.
-
-This module creates an HTTP server that exposes the PythonEnvironment
-over HTTP and WebSocket endpoints, compatible with EnvClient.
-
-Endpoints:
-    - POST /reset: Reset the environment
-    - POST /step: Execute an action
-    - GET /state: Get current environment state
-    - GET /schema: Get action/observation schemas
-    - WS /ws: WebSocket endpoint for persistent sessions
-
-Usage:
-    # Development (with auto-reload):
-    uvicorn server.app:app --reload --host 0.0.0.0 --port 8000
-
-    # Production:
-    uvicorn server.app:app --host 0.0.0.0 --port 8000 --workers 4
-
-    # Or run directly:
-    python -m server.app
-"""
+from __future__ import annotations
 
 try:
     from openenv.core.env_server.http_server import create_app
-except Exception as e:  # pragma: no cover
+except Exception as exc:  # pragma: no cover
     raise ImportError(
-        "openenv is required for the web interface. Install dependencies with '\n    uv sync\n'"
-    ) from e
+        "openenv-core is required to run the API server. Install project dependencies first."
+    ) from exc
 
 try:
-    from ..models import PythonAction, PythonObservation
-    from .python_env_environment import PythonEnvironment
-except ModuleNotFoundError:
-    from models import PythonAction, PythonObservation
-    from server.python_env_environment import PythonEnvironment
+    from ..models import PythonCodeReviewAction, PythonCodeReviewObservation
+    from .env import PythonCodeReviewEnvironment
+except ImportError:
+    from models import PythonCodeReviewAction, PythonCodeReviewObservation
+    from server.env import PythonCodeReviewEnvironment
 
 
-# Create the app with web interface and README integration
 app = create_app(
-    PythonEnvironment,
-    PythonAction,
-    PythonObservation,
-    env_name="python_env",
-    max_concurrent_envs=1,  # increase this number to allow more concurrent WebSocket sessions
+    PythonCodeReviewEnvironment,
+    PythonCodeReviewAction,
+    PythonCodeReviewObservation,
+    env_name="python_code_review_env",
+    max_concurrent_envs=4,
 )
 
 
-def main(host: str = "0.0.0.0", port: int = 8000):
-    """
-    Entry point for direct execution via uv run or python -m.
-
-    This function enables running the server without Docker:
-        uv run --project . server
-        uv run --project . server --port 8001
-        python -m python_env.server.app
-
-    Args:
-        host: Host address to bind to (default: "0.0.0.0")
-        port: Port number to listen on (default: 8000)
-
-    For production deployments, consider using uvicorn directly with
-    multiple workers:
-        uvicorn python_env.server.app:app --workers 4
-    """
+def main(host: str = "0.0.0.0", port: int = 8000) -> None:
     import uvicorn
 
     uvicorn.run(app, host=host, port=port)
 
 
 if __name__ == "__main__":
-    import argparse
-
-    parser = argparse.ArgumentParser()
-    parser.add_argument("--port", type=int, default=8000)
-    args = parser.parse_args()
-    main(port=args.port)
+    main()

server/env.py

@@ -0,0 +1,381 @@
+"""OpenEnv environment implementation for Python code review tasks."""
+
+from __future__ import annotations
+
+from typing import Any, Dict, Optional, Tuple
+from uuid import uuid4
+
+from openenv.core.env_server.interfaces import Environment
+from openenv.core.env_server.types import EnvironmentMetadata
+
+try:
+    from ..graders import grade_task
+    from ..models import (
+        HistoryEntry,
+        PythonCodeReviewAction,
+        PythonCodeReviewObservation,
+        PythonCodeReviewState,
+        RewardDetails,
+        TaskGrade,
+    )
+    from ..tasks import ReviewTask, list_tasks, select_task
+except ImportError:
+    from graders import grade_task
+    from models import (
+        HistoryEntry,
+        PythonCodeReviewAction,
+        PythonCodeReviewObservation,
+        PythonCodeReviewState,
+        RewardDetails,
+        TaskGrade,
+    )
+    from tasks import ReviewTask, list_tasks, select_task
+
+
+def _empty_grade() -> TaskGrade:
+    return TaskGrade(score=0.0, syntax_score=0.0, tests_passed=0, tests_total=0, quality_score=0.0, runtime_score=0.0)
+
+
+def _clamp(value: float, lower: float = -1.0, upper: float = 1.0) -> float:
+    return max(lower, min(upper, value))
+
+
+class PythonCodeReviewEnvironment(
+    Environment[PythonCodeReviewAction, PythonCodeReviewObservation, PythonCodeReviewState]
+):
+    """Structured environment for deterministic Python code review workflows."""
+
+    SUPPORTS_CONCURRENT_SESSIONS: bool = True
+
+    def __init__(self, verbose: bool = False, **_: Any) -> None:
+        super().__init__()
+        self.verbose = verbose
+        self._task: ReviewTask = list_tasks()[0]
+        self._current_code: str = self._task.starter_code
+        self._history: list[HistoryEntry] = []
+        self._last_reward = RewardDetails(value=0.0, reason="Environment initialized.")
+        self._current_grade = _empty_grade()
+        self._state = PythonCodeReviewState(episode_id=str(uuid4()), step_count=0)
+        self.reset()
+
+    def reset(
+        self,
+        seed: Optional[int] = None,
+        episode_id: Optional[str] = None,
+        **kwargs: Any,
+    ) -> PythonCodeReviewObservation:
+        task_id = kwargs.get("task_id")
+        self._task = select_task(seed=seed, task_id=task_id)
+        self._current_code = self._task.starter_code
+        self._history = []
+        self._last_reward = RewardDetails(value=0.0, reason="Environment reset.")
+        self._current_grade = grade_task(self._task, self._current_code, include_hidden=False)
+
+        self._state = PythonCodeReviewState(
+            episode_id=episode_id or str(uuid4()),
+            step_count=0,
+            task_id=self._task.task_id,
+            difficulty=self._task.difficulty,
+            task_kind=self._task.task_kind,
+            attempts_remaining=self._task.max_steps,
+            current_code=self._current_code,
+            errors=self._format_errors(self._current_grade),
+            test_results=self._format_test_results(self._current_grade),
+            history=[],
+            score=self._current_grade.score,
+            done=False,
+        )
+        return self._build_observation(
+            grade=self._current_grade,
+            status=f"Loaded task {self._task.task_id}.",
+            reward_details=self._last_reward,
+        )
+
+    def step(
+        self,
+        action: PythonCodeReviewAction,
+        timeout_s: Optional[float] = None,
+        **kwargs: Any,
+    ) -> PythonCodeReviewObservation:
+        observation, _, _, _ = self._step_transition(action, timeout_s=timeout_s, **kwargs)
+        return observation
+
+    def step_result(
+        self,
+        action: PythonCodeReviewAction,
+        timeout_s: Optional[float] = None,
+        **kwargs: Any,
+    ) -> Tuple[PythonCodeReviewObservation, float, bool, Dict[str, Any]]:
+        """Gym-style helper used by local scripts and tests."""
+
+        return self._step_transition(action, timeout_s=timeout_s, **kwargs)
+
+    def _step_transition(
+        self,
+        action: PythonCodeReviewAction,
+        timeout_s: Optional[float] = None,
+        **kwargs: Any,
+    ) -> Tuple[PythonCodeReviewObservation, float, bool, Dict[str, Any]]:
+        if self._state.done:
+            reward = RewardDetails(value=0.0, reason="Episode already finished. Call reset() to continue.")
+            observation = self._build_observation(
+                grade=self._current_grade,
+                status="Episode already finished.",
+                reward_details=reward,
+            )
+            return observation, reward.value, observation.done, {"task_id": observation.task_id, "score": observation.score}
+
+        previous_grade = self._current_grade
+        status = ""
+        invalid_action = False
+        code_changed = False
+        use_hidden_grading = False
+
+        if action.action_type == "edit_code":
+            if not action.code or not action.code.strip():
+                invalid_action = True
+                status = "edit_code requires a non-empty code payload."
+            else:
+                code_changed = action.code != self._current_code
+                self._current_code = action.code
+                status = "Updated working copy from agent patch."
+        elif action.action_type == "submit_solution":
+            if action.code is not None and action.code.strip():
+                code_changed = action.code != self._current_code
+                self._current_code = action.code
+            use_hidden_grading = True
+            status = "Submission received for final grading."
+        elif action.action_type == "run_tests":
+            status = "Executed public validation suite."
+        elif action.action_type == "analyze_code":
+            status = "Generated static review summary."
+        else:  # pragma: no cover
+            invalid_action = True
+            status = f"Unsupported action_type: {action.action_type}"
+
+        self._state.step_count += 1
+
+        if invalid_action:
+            current_grade = previous_grade
+        else:
+            current_grade = grade_task(
+                self._task,
+                self._current_code,
+                include_hidden=use_hidden_grading,
+                timeout_s=timeout_s or 3.0,
+            )
+            if action.action_type == "analyze_code":
+                status = self._analysis_status(current_grade)
+            elif action.action_type == "run_tests":
+                status = self._run_tests_status(current_grade, use_hidden_grading)
+            elif action.action_type == "submit_solution":
+                status = self._submission_status(current_grade)
+
+        done = use_hidden_grading or self._state.step_count >= self._task.max_steps
+        if self._state.step_count >= self._task.max_steps and not use_hidden_grading:
+            status = f"{status} Step budget exhausted."
+
+        reward_details = self._compute_reward(
+            previous_grade=previous_grade,
+            current_grade=current_grade,
+            action=action,
+            invalid_action=invalid_action,
+            timed_out=current_grade.timed_out,
+            code_changed=code_changed,
+            final_submission=use_hidden_grading,
+        )
+
+        self._history.append(
+            HistoryEntry(
+                step=self._state.step_count,
+                action_type=action.action_type,
+                status=status,
+                reward=reward_details.value,
+            )
+        )
+
+        self._current_grade = current_grade
+        self._last_reward = reward_details
+        attempts_remaining = max(self._task.max_steps - self._state.step_count, 0)
+
+        self._state.task_id = self._task.task_id
+        self._state.difficulty = self._task.difficulty
+        self._state.task_kind = self._task.task_kind
+        self._state.attempts_remaining = attempts_remaining
+        self._state.current_code = self._current_code
+        self._state.errors = self._format_errors(current_grade)
+        self._state.test_results = self._format_test_results(current_grade)
+        self._state.history = list(self._history)
+        self._state.score = current_grade.score
+        self._state.done = done
+
+        observation = self._build_observation(
+            grade=current_grade,
+            status=status,
+            reward_details=reward_details,
+        )
+        return observation, reward_details.value, observation.done, {"task_id": observation.task_id, "score": observation.score}
+
+    @property
+    def state(self) -> PythonCodeReviewState:
+        return self._state
+
+    def _build_observation(
+        self,
+        *,
+        grade: TaskGrade,
+        status: str,
+        reward_details: RewardDetails,
+    ) -> PythonCodeReviewObservation:
+        return PythonCodeReviewObservation(
+            task_id=self._task.task_id,
+            title=self._task.title,
+            difficulty=self._task.difficulty,
+            task_kind=self._task.task_kind,
+            task_description=self._task.task_description,
+            current_code=self._current_code,
+            errors=self._format_errors(grade),
+            test_results=self._format_test_results(grade),
+            visible_tests=list(self._task.visible_tests),
+            history=list(self._history),
+            attempts_remaining=self._state.attempts_remaining,
+            last_action_status=status,
+            score=grade.score,
+            reward=reward_details.value,
+            done=self._state.done,
+            reward_details=reward_details,
+            metadata={
+                "goal": self._task.goal,
+                "repo_summary": self._task.repo_summary,
+                "changed_files": self._task.changed_files,
+                "available_files": self._task.available_files,
+                "grade_details": grade.details,
+            },
+        )
+
+    def _compute_reward(
+        self,
+        *,
+        previous_grade: TaskGrade,
+        current_grade: TaskGrade,
+        action: PythonCodeReviewAction,
+        invalid_action: bool,
+        timed_out: bool,
+        code_changed: bool,
+        final_submission: bool,
+    ) -> RewardDetails:
+        prev_score = previous_grade.score
+        curr_score = current_grade.score
+        prev_rate = previous_grade.tests_passed / max(previous_grade.tests_total, 1)
+        curr_rate = current_grade.tests_passed / max(current_grade.tests_total, 1)
+
+        syntax_reward = 0.2 if previous_grade.syntax_score < 1.0 and current_grade.syntax_score >= 1.0 else 0.0
+        test_reward = round(max(curr_rate - prev_rate, 0.0) * 0.3, 3)
+        progress_delta = round(max(curr_score - prev_score, 0.0) * 0.4, 3)
+        quality_bonus = round(max(current_grade.quality_score - previous_grade.quality_score, 0.0) * 0.1, 3)
+        correctness_bonus = 0.5 if final_submission and curr_score >= 0.999 and prev_score < 0.999 else 0.0
+
+        invalid_action_penalty = 0.1 if invalid_action else 0.0
+        timeout_penalty = 0.2 if timed_out else 0.0
+        regression_penalty = round(max(prev_score - curr_score, 0.0) * 0.2, 3)
+        stagnation_penalty = 0.05 if action.action_type == "edit_code" and not code_changed else 0.0
+
+        value = _clamp(
+            syntax_reward
+            + test_reward
+            + progress_delta
+            + quality_bonus
+            + correctness_bonus
+            - invalid_action_penalty
+            - timeout_penalty
+            - regression_penalty
+            - stagnation_penalty
+        )
+
+        reason_parts = []
+        if syntax_reward:
+            reason_parts.append("syntax fixed")
+        if test_reward:
+            reason_parts.append("public test progress")
+        if progress_delta:
+            reason_parts.append("overall score improved")
+        if quality_bonus:
+            reason_parts.append("code quality improved")
+        if correctness_bonus:
+            reason_parts.append("full correctness bonus")
+        if invalid_action_penalty:
+            reason_parts.append("invalid action penalty")
+        if timeout_penalty:
+            reason_parts.append("timeout penalty")
+        if regression_penalty:
+            reason_parts.append("regression penalty")
+        if stagnation_penalty:
+            reason_parts.append("unchanged patch penalty")
+        if not reason_parts:
+            reason_parts.append("no meaningful state change")
+
+        return RewardDetails(
+            value=round(value, 3),
+            syntax_reward=syntax_reward,
+            test_reward=test_reward,
+            correctness_bonus=correctness_bonus,
+            quality_bonus=quality_bonus,
+            progress_delta=progress_delta,
+            invalid_action_penalty=invalid_action_penalty,
+            timeout_penalty=timeout_penalty,
+            regression_penalty=regression_penalty,
+            stagnation_penalty=stagnation_penalty,
+            reason=", ".join(reason_parts),
+            prev_score=prev_score,
+            curr_score=curr_score,
+            code_changed=code_changed,
+        )
+
+    def _format_errors(self, grade: TaskGrade) -> str:
+        compile_error = str(grade.details.get("compile_error", "")).strip()
+        if compile_error:
+            return compile_error
+        return "Code parses successfully."
+
+    def _format_test_results(self, grade: TaskGrade) -> str:
+        parts = [grade.details.get("test_summary", "No test feedback available.")]
+        benchmark = grade.details.get("benchmark")
+        if isinstance(benchmark, dict):
+            parts.append(
+                "Benchmark: "
+                f"candidate {benchmark['candidate_seconds']}s vs baseline {benchmark['baseline_seconds']}s "
+                f"(x{benchmark['improvement_ratio']})."
+            )
+        elif isinstance(benchmark, str) and benchmark:
+            parts.append(f"Benchmark: {benchmark}")
+        return "\n".join(part for part in parts if part)
+
+    def _analysis_status(self, grade: TaskGrade) -> str:
+        notes = grade.details.get("quality_notes", [])
+        quality_note = notes[0] if notes else "No major static quality issues detected."
+        return (
+            f"Syntax score {grade.syntax_score:.2f}; "
+            f"public tests {grade.tests_passed}/{grade.tests_total}; "
+            f"quality {grade.quality_score:.2f}. {quality_note}"
+        )
+
+    def _run_tests_status(self, grade: TaskGrade, include_hidden: bool) -> str:
+        visibility = "full" if include_hidden else "public"
+        return f"Ran {visibility} tests: {grade.tests_passed}/{grade.tests_total} passed."
+
+    def _submission_status(self, grade: TaskGrade) -> str:
+        runtime_text = ""
+        if grade.runtime_score:
+            runtime_text = f" runtime {grade.runtime_score:.2f};"
+        return (
+            f"Submission graded with score {grade.score:.2f}; "
+            f"tests {grade.tests_passed}/{grade.tests_total};"
+            f"{runtime_text} quality {grade.quality_score:.2f}."
+        )
+
+    def get_metadata(self) -> EnvironmentMetadata:
+        return EnvironmentMetadata(
+            name="python_code_review_env",
+            description="Production-style Python code review environment with deterministic grading.",
+            version="1.0.0",
+        )

server/python_env_environment.py

@@ -1,104 +1,3 @@
-# Copyright (c) Meta Platforms, Inc. and affiliates.
-# All rights reserved.
-#
-# This source code is licensed under the BSD-style license found in the
-# LICENSE file in the root directory of this source tree.
+"""Backward-compatible import shim for the environment class."""
 
-"""
-Python Env Environment Implementation.
-
-A simple test environment that echoes back messages sent to it.
-Perfect for testing HTTP server infrastructure.
-"""
-
-from uuid import uuid4
-
-from openenv.core.env_server.interfaces import Environment
-from openenv.core.env_server.types import State
-
-try:
-    from ..models import PythonAction, PythonObservation
-except ImportError:
-    from models import PythonAction, PythonObservation
-
-
-class PythonEnvironment(Environment):
-    """
-    A simple echo environment that echoes back messages.
-
-    This environment is designed for testing the HTTP server infrastructure.
-    It maintains minimal state and simply echoes back whatever message it receives.
-
-    Example:
-        >>> env = PythonEnvironment()
-        >>> obs = env.reset()
-        >>> print(obs.echoed_message)  # "Python Env environment ready!"
-        >>>
-        >>> obs = env.step(PythonAction(message="Hello"))
-        >>> print(obs.echoed_message)  # "Hello"
-        >>> print(obs.message_length)  # 5
-    """
-
-    # Enable concurrent WebSocket sessions.
-    # Set to True if your environment isolates state between instances.
-    # When True, multiple WebSocket clients can connect simultaneously, each
-    # getting their own environment instance (when using factory mode in app.py).
-    SUPPORTS_CONCURRENT_SESSIONS: bool = True
-
-    def __init__(self):
-        """Initialize the python_env environment."""
-        self._state = State(episode_id=str(uuid4()), step_count=0)
-        self._reset_count = 0
-
-    def reset(self) -> PythonObservation:
-        """
-        Reset the environment.
-
-        Returns:
-            PythonObservation with a ready message
-        """
-        self._state = State(episode_id=str(uuid4()), step_count=0)
-        self._reset_count += 1
-
-        return PythonObservation(
-            echoed_message="Python Env environment ready!",
-            message_length=0,
-            done=False,
-            reward=0.0,
-        )
-
-    def step(self, action: PythonAction) -> PythonObservation:  # type: ignore[override]
-        """
-        Execute a step in the environment by echoing the message.
-
-        Args:
-            action: PythonAction containing the message to echo
-
-        Returns:
-            PythonObservation with the echoed message and its length
-        """
-        self._state.step_count += 1
-
-        message = action.message
-        length = len(message)
-
-        # Simple reward: longer messages get higher rewards
-        reward = length * 0.1
-
-        return PythonObservation(
-            echoed_message=message,
-            message_length=length,
-            done=False,
-            reward=reward,
-            metadata={"original_message": message, "step": self._state.step_count},
-        )
-
-    @property
-    def state(self) -> State:
-        """
-        Get the current environment state.
-
-        Returns:
-            Current State with episode_id and step_count
-        """
-        return self._state
+from .env import PythonCodeReviewEnvironment, PythonCodeReviewEnvironment as PythonEnvironment

server/requirements.txt

@@ -1,6 +1,5 @@
-openenv[core]>=0.2.0
-fastapi>=0.115.0
-uvicorn>=0.24.0
-
-
-
+openenv-core[core]>=0.2.2
+fastapi>=0.111.0
+uvicorn>=0.30.0
+pytest>=8.0.0
+openai>=1.76.0

tasks/__init__.py

@@ -0,0 +1,12 @@
+"""Task catalog for python_code_review_env."""
+
+from .catalog import ReviewTask, get_task, list_tasks, select_task
+
+
+def task_ids() -> list[str]:
+    """Return stable task identifiers for validators."""
+
+    return [task.task_id for task in list_tasks()]
+
+
+__all__ = ["ReviewTask", "get_task", "list_tasks", "select_task", "task_ids"]

tasks/catalog.py

@@ -0,0 +1,324 @@
+"""Deterministic task definitions for the code review environment."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from textwrap import dedent
+from typing import Any, Dict, List, Optional
+
+
+def _code(value: str) -> str:
+    return dedent(value).strip() + "\n"
+
+
+@dataclass(frozen=True)
+class CallCase:
+    """One executable function call used by graders."""
+
+    label: str
+    args: tuple[Any, ...] = ()
+    kwargs: Dict[str, Any] = field(default_factory=dict)
+    expected: Any = None
+
+
+@dataclass(frozen=True)
+class ReviewTask:
+    """Static task definition."""
+
+    task_id: str
+    title: str
+    difficulty: str
+    task_kind: str
+    task_description: str
+    starter_code: str
+    reference_code: str
+    function_name: str
+    visible_tests: List[str]
+    public_cases: List[CallCase]
+    hidden_cases: List[CallCase]
+    repo_summary: str
+    changed_files: List[str]
+    available_files: List[str]
+    goal: str
+    max_steps: int
+    benchmark_config: Optional[Dict[str, int]] = None
+
+
+TASKS: List[ReviewTask] = [
+    ReviewTask(
+        task_id="syntax_fix_invoice_totals",
+        title="Fix the invoice total syntax regression",
+        difficulty="easy",
+        task_kind="syntax_fix",
+        task_description=(
+            "A recent refactor broke the helper that normalizes invoice totals before "
+            "daily reconciliation. Repair the Python syntax so the function compiles "
+            "and returns the correct total for mixed integer and string inputs."
+        ),
+        starter_code=_code(
+            """
+            def normalize_invoice_totals(records):
+                cleaned = []
+                for record in records
+                    if "total" not in record:
+                        continue
+                    value = int(record["total"])
+                    cleaned.append(value)
+                return sum(cleaned
+            """
+        ),
+        reference_code=_code(
+            '''
+            def normalize_invoice_totals(records):
+                """Return the sum of invoice totals that are present in the payload."""
+                cleaned = []
+                for record in records:
+                    if "total" not in record:
+                        continue
+                    cleaned.append(int(record["total"]))
+                return sum(cleaned)
+            '''
+        ),
+        function_name="normalize_invoice_totals",
+        visible_tests=[
+            "normalize_invoice_totals([{'total': '4'}, {'total': 5}, {}]) == 9",
+            "normalize_invoice_totals([]) == 0",
+        ],
+        public_cases=[
+            CallCase(
+                label="mixed string and int totals",
+                args=([{"total": "4"}, {"total": 5}, {}],),
+                expected=9,
+            ),
+            CallCase(label="empty input", args=([],), expected=0),
+        ],
+        hidden_cases=[
+            CallCase(
+                label="skip missing totals",
+                args=([{}, {"total": "2"}, {"total": "8"}],),
+                expected=10,
+            ),
+            CallCase(
+                label="handle negative adjustments",
+                args=([{"total": "11"}, {"total": -3}],),
+                expected=8,
+            ),
+        ],
+        repo_summary=(
+            "services/billing/reconciliation.py computes end-of-day invoice totals for "
+            "a CPU-only batch job."
+        ),
+        changed_files=["services/billing/reconciliation.py"],
+        available_files=["services/billing/reconciliation.py", "tests/test_reconciliation.py"],
+        goal="Restore a compiling implementation for invoice total normalization.",
+        max_steps=6,
+    ),
+    ReviewTask(
+        task_id="bug_fix_session_windows",
+        title="Repair session window collapsing logic",
+        difficulty="medium",
+        task_kind="bug_fix",
+        task_description=(
+            "The session aggregator regressed after a cleanup pass. Public tests expose "
+            "incorrect boundary handling and the final session is missing. Fix the logic "
+            "without changing the function contract."
+        ),
+        starter_code=_code(
+            """
+            def collapse_sessions(events, idle_timeout_minutes):
+                if not events:
+                    return []
+
+                sessions = []
+                current_start = events[0]["minute"]
+                current_end = current_start
+
+                for event in events[1:]:
+                    minute = event["minute"]
+                    if minute - current_end > idle_timeout_minutes:
+                        sessions.append((current_start, current_end))
+                        current_start = minute
+                    current_end = minute
+
+                return sessions
+            """
+        ),
+        reference_code=_code(
+            '''
+            def collapse_sessions(events, idle_timeout_minutes):
+                """Collapse activity events into inclusive session windows."""
+                if not events:
+                    return []
+
+                sessions = []
+                current_start = events[0]["minute"]
+                current_end = current_start
+
+                for event in events[1:]:
+                    minute = event["minute"]
+                    if minute - current_end >= idle_timeout_minutes:
+                        sessions.append((current_start, current_end))
+                        current_start = minute
+                    current_end = minute
+
+                sessions.append((current_start, current_end))
+                return sessions
+            '''
+        ),
+        function_name="collapse_sessions",
+        visible_tests=[
+            "collapse_sessions([{'minute': 1}, {'minute': 3}, {'minute': 8}], 4) == [(1, 3), (8, 8)]",
+            "collapse_sessions([{'minute': 5}, {'minute': 9}], 4) == [(5, 5), (9, 9)]",
+        ],
+        public_cases=[
+            CallCase(
+                label="split when idle timeout is exceeded",
+                args=([{"minute": 1}, {"minute": 3}, {"minute": 8}], 4),
+                expected=[(1, 3), (8, 8)],
+            ),
+            CallCase(
+                label="boundary is inclusive",
+                args=([{"minute": 5}, {"minute": 9}], 4),
+                expected=[(5, 5), (9, 9)],
+            ),
+        ],
+        hidden_cases=[
+            CallCase(
+                label="single continuous session",
+                args=([{"minute": 2}, {"minute": 4}, {"minute": 5}], 4),
+                expected=[(2, 5)],
+            ),
+            CallCase(label="empty input", args=([], 10), expected=[]),
+            CallCase(
+                label="multiple boundaries",
+                args=([{"minute": 1}, {"minute": 5}, {"minute": 9}, {"minute": 14}], 4),
+                expected=[(1, 1), (5, 5), (9, 9), (14, 14)],
+            ),
+        ],
+        repo_summary=(
+            "analytics/sessionizer.py condenses sorted clickstream events into user "
+            "sessions for downstream retention reports."
+        ),
+        changed_files=["analytics/sessionizer.py"],
+        available_files=["analytics/sessionizer.py", "tests/test_sessionizer.py"],
+        goal="Make session collapsing match the expected timeout semantics.",
+        max_steps=8,
+    ),
+    ReviewTask(
+        task_id="optimization_rank_active_users",
+        title="Optimize the active-user ranking pipeline",
+        difficulty="hard",
+        task_kind="optimization",
+        task_description=(
+            "The reporting job is correct enough for small fixtures but too slow for the "
+            "daily production export. Preserve the API, keep the output deterministic, "
+            "and refactor the implementation for speed and readability."
+        ),
+        starter_code=_code(
+            """
+            def rank_active_users(events):
+                users = []
+                for event in events:
+                    if event["status"] == "active":
+                        found = False
+                        for existing in users:
+                            if existing == event["user_id"]:
+                                found = True
+                        if not found:
+                            users.append(event["user_id"])
+
+                totals = []
+                for user in users:
+                    count = 0
+                    for event in events:
+                        if event["status"] == "active" and event["user_id"] == user:
+                            count = count + 1
+                    totals.append((user, count))
+
+                totals.sort(key=lambda item: (-item[1], item[0]))
+                return totals
+            """
+        ),
+        reference_code=_code(
+            '''
+            from collections import Counter
+
+
+            def rank_active_users(events):
+                """Return users ranked by number of active events."""
+                counts = Counter(
+                    event["user_id"]
+                    for event in events
+                    if event["status"] == "active"
+                )
+                return sorted(counts.items(), key=lambda item: (-item[1], item[0]))
+            '''
+        ),
+        function_name="rank_active_users",
+        visible_tests=[
+            "rank_active_users([{'user_id': 'b', 'status': 'active'}, {'user_id': 'a', 'status': 'active'}, {'user_id': 'b', 'status': 'inactive'}]) == [('a', 1), ('b', 1)]",
+            "rank_active_users([{'user_id': 'u1', 'status': 'active'}, {'user_id': 'u1', 'status': 'active'}, {'user_id': 'u2', 'status': 'active'}]) == [('u1', 2), ('u2', 1)]",
+        ],
+        public_cases=[
+            CallCase(
+                label="inactive events are ignored",
+                args=([{"user_id": "b", "status": "active"}, {"user_id": "a", "status": "active"}, {"user_id": "b", "status": "inactive"}],),
+                expected=[("a", 1), ("b", 1)],
+            ),
+            CallCase(
+                label="counts repeated active users",
+                args=([{"user_id": "u1", "status": "active"}, {"user_id": "u1", "status": "active"}, {"user_id": "u2", "status": "active"}],),
+                expected=[("u1", 2), ("u2", 1)],
+            ),
+        ],
+        hidden_cases=[
+            CallCase(
+                label="stable alphabetical tie-break",
+                args=([{"user_id": "u3", "status": "active"}, {"user_id": "u2", "status": "active"}, {"user_id": "u3", "status": "active"}, {"user_id": "u2", "status": "active"}],),
+                expected=[("u2", 2), ("u3", 2)],
+            ),
+            CallCase(label="empty input", args=([],), expected=[]),
+            CallCase(
+                label="mixed active and inactive states",
+                args=([{"user_id": "x", "status": "inactive"}, {"user_id": "x", "status": "active"}, {"user_id": "y", "status": "active"}, {"user_id": "x", "status": "active"}],),
+                expected=[("x", 2), ("y", 1)],
+            ),
+        ],
+        repo_summary=(
+            "reports/activity_rankings.py feeds a nightly export that runs on a small CPU "
+            "instance and has become too slow after customer growth."
+        ),
+        changed_files=["reports/activity_rankings.py"],
+        available_files=["reports/activity_rankings.py", "tests/test_activity_rankings.py"],
+        goal="Keep the output stable while improving runtime and code quality.",
+        max_steps=10,
+        benchmark_config={"user_pool": 240, "events_per_user": 36, "iterations": 8},
+    ),
+]
+
+TASK_BY_ID = {task.task_id: task for task in TASKS}
+
+
+def list_tasks() -> List[ReviewTask]:
+    """Return all supported tasks."""
+
+    return list(TASKS)
+
+
+def get_task(task_id: str) -> ReviewTask:
+    """Fetch a task by identifier."""
+
+    try:
+        return TASK_BY_ID[task_id]
+    except KeyError as exc:
+        raise ValueError(f"Unknown task_id: {task_id}") from exc
+
+
+def select_task(seed: Optional[int] = None, task_id: Optional[str] = None) -> ReviewTask:
+    """Select a task deterministically by explicit id or seed."""
+
+    if task_id:
+        return get_task(task_id)
+    if seed is None:
+        return TASKS[0]
+    return TASKS[seed % len(TASKS)]