Upload folder using huggingface_hub

Dockerfile

@@ -0,0 +1,23 @@
+FROM python:3.11-slim
+
+ENV PYTHONDONTWRITEBYTECODE=1 \
+    PYTHONUNBUFFERED=1 \
+    PIP_NO_CACHE_DIR=1
+
+WORKDIR /app
+
+COPY pyproject.toml README.md openenv.yaml __init__.py client.py compat.py models.py inference.py /app/
+COPY server /app/server
+COPY tasks /app/tasks
+COPY graders /app/graders
+
+RUN python -m pip install --upgrade pip && \
+    pip install .
+
+EXPOSE 8000
+
+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()"
+
+ENV ENABLE_WEB_INTERFACE=true
+CMD ["uvicorn", "server.app:app", "--host", "0.0.0.0", "--port", "8000"]

README.md

@@ -1,10 +1,189 @@
----
-title: Final Python Env
-emoji: 🐠
-colorFrom: green
-colorTo: blue
-sdk: docker
-pinned: false
----
-
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+---
+title: Python Code Review Environment
+colorFrom: yellow
+colorTo: blue
+sdk: docker
+pinned: false
+app_port: 8000
+tags:
+  - openenv
+  - code-review
+  - python
+base_path: /web
+---
+
+# python_code_review_env
+
+`python_code_review_env` is a production-style OpenEnv environment that simulates a realistic Python code review workflow. An agent inspects broken code, edits it, runs tests, and submits a final solution against deterministic graders for syntax repair, bug fixing, and optimization/refactoring.
+
+## Environment design
+
+- `Observation` includes task instructions, current code, syntax errors, public test output, action history, and remaining attempts.
+- `Action` is structured as `analyze_code`, `edit_code`, `run_tests`, or `submit_solution`.
+- `Reward` is shaped and non-binary. The environment awards syntax progress, test progress, correctness, and quality improvements while penalizing invalid actions, timeouts, regressions, and unchanged edits.
+- `State` exposes the internal episode snapshot through `/state`.
+
+## Task set
+
+1. `syntax_fix_invoice_totals` (easy)
+   Fix a syntax regression in an invoice normalization helper.
+2. `bug_fix_session_windows` (medium)
+   Repair a session-collapsing bug using deterministic public and hidden tests.
+3. `optimization_rank_active_users` (hard)
+   Refactor a slow ranking function and earn additional score from runtime improvement plus AST/style quality.
+
+## Action schema
+
+```json
+{
+  "action_type": "edit_code",
+  "code": "def function(...):\n    ..."
+}
+```
+
+Supported `action_type` values:
+
+- `analyze_code`
+- `edit_code`
+- `run_tests`
+- `submit_solution`
+
+## Observation schema
+
+```json
+{
+  "task_description": "...",
+  "current_code": "...",
+  "errors": "...",
+  "test_results": "...",
+  "history": []
+}
+```
+
+The full observation also includes `task_id`, `difficulty`, `task_kind`, `visible_tests`, `attempts_remaining`, `score`, `last_action_status`, `reward`, `done`, and a structured `reward_details` breakdown.
+
+## Deterministic grading
+
+- Syntax tasks use `compile()` plus hidden behavioral checks.
+- Bug-fix tasks use deterministic function-call cases that behave like pytest assertions.
+- Optimization tasks combine correctness, runtime benchmarking, and AST/style quality scoring.
+- Infinite loops and long-running solutions are sandboxed with subprocess timeouts and receive penalties.
+- All scores are clamped to `[0.0, 1.0]`.
+
+## Run locally
+
+Install dependencies:
+
+```bash
+pip install .
+```
+
+Start the API server:
+
+```bash
+uvicorn server.app:app --host 0.0.0.0 --port 8000
+```
+
+Smoke-test the environment:
+
+```bash
+curl http://localhost:8000/health
+curl http://localhost:8000/state
+```
+
+OpenEnv validation:
+
+```bash
+openenv validate
+```
+
+## Docker build
+
+The Docker image no longer depends on `ghcr.io/meta-pytorch/openenv-base:latest`, which removes the TLS handshake failure from the original build path.
+
+```bash
+# Run from repo root
+docker build -t python-code-review-env -f server/Dockerfile .
+docker run --rm -p 8000:8000 python-code-review-env
+```
+
+If you run the build from inside `server/`, you must point the context at the repo root:
+
+```bash
+docker build -t python-code-review-env -f Dockerfile ..
+```
+
+Expected health check:
+
+```bash
+curl http://localhost:8000/health
+```
+
+## Hugging Face Spaces deployment
+
+1. Create a Docker Space.
+2. Push this repository content to the Space.
+3. Ensure port `8000` is exposed.
+4. Wait for the container to build.
+5. Verify `/reset` and `/health` return `200`.
+
+The image is CPU-friendly and designed for a small Hugging Face Space such as `2 vCPU / 8 GB RAM`.
+
+## Inference baseline
+
+`inference.py` uses an OpenAI-compatible client:
+
+```python
+client = OpenAI(base_url=API_BASE_URL, api_key=API_KEY)
+```
+
+Supported providers include:
+
+- Gemini through an OpenAI-compatible gateway
+- OpenRouter
+- Together AI
+- DeepSeek-compatible OpenAI endpoints
+
+Run it with a free/open provider:
+
+```bash
+set API_BASE_URL=https://openrouter.ai/api/v1
+set API_KEY=...
+set MODEL=deepseek/deepseek-chat-v3-0324:free
+python inference.py
+```
+
+If no credentials are supplied, the script falls back to a deterministic smoke-test policy that applies the reference fix for each task so the environment can still be validated end to end.
+
+Example output:
+
+```text
+Task 1 Score: 1.0
+Task 2 Score: 1.0
+Task 3 Score: 0.9
+Final Score: 1.0
+```
+
+## Project structure
+
+```text
+python_env/
+├── client.py
+├── graders/
+│   ├── bug_fix.py
+│   ├── dispatch.py
+│   ├── optimization.py
+│   ├── shared.py
+│   └── syntax.py
+├── inference.py
+├── models.py
+├── openenv.yaml
+├── README.md
+├── server/
+│   ├── app.py
+│   ├── Dockerfile
+│   ├── env.py
+│   └── python_env_environment.py
+└── tasks/
+    └── catalog.py
+```

__init__.py

@@ -0,0 +1,22 @@
+"""Public package exports for python_code_review_env."""
+
+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

@@ -0,0 +1,37 @@
+"""Client helpers for python_code_review_env."""
+
+from __future__ import annotations
+
+from typing import Dict
+
+from openenv.core import EnvClient
+from openenv.core.client_types import StepResult
+
+from .models import (
+    PythonCodeReviewAction,
+    PythonCodeReviewObservation,
+    PythonCodeReviewState,
+)
+
+
+class PythonCodeReviewEnv(
+    EnvClient[PythonCodeReviewAction, PythonCodeReviewObservation, PythonCodeReviewState]
+):
+    """Typed client for the code review environment."""
+
+    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", observation.done),
+        )
+
+    def _parse_state(self, payload: Dict) -> PythonCodeReviewState:
+        return PythonCodeReviewState.model_validate(payload)
+
+
+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

@@ -0,0 +1,140 @@
+"""Typed models for the python_code_review_env environment."""
+
+from __future__ import annotations
+
+from typing import Any, Dict, List, Literal, Optional
+
+from pydantic import BaseModel, Field
+
+from openenv.core.env_server.types import Action, Observation, State
+
+
+Difficulty = Literal["easy", "medium", "hard"]
+TaskKind = Literal["syntax_fix", "bug_fix", "optimization"]
+ActionType = Literal["analyze_code", "edit_code", "run_tests", "submit_solution"]
+
+
+class HistoryEntry(BaseModel):
+    """One environment transition recorded for the agent."""
+
+    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

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

pyproject.toml

@@ -0,0 +1,35 @@
+[build-system]
+requires = ["setuptools>=68", "wheel"]
+build-backend = "setuptools.build_meta"
+
+[project]
+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 = [
+    "fastapi>=0.111.0",
+    "openai>=1.76.0",
+    "openenv-core[core]>=0.2.2",
+    "pytest>=8.0.0",
+    "uvicorn>=0.30.0",
+]
+
+[project.optional-dependencies]
+dev = [
+    "pytest-cov>=4.0.0",
+]
+
+[project.scripts]
+server = "python_env.server.app:main"
+
+[tool.setuptools]
+include-package-data = true
+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/__init__.py

@@ -0,0 +1,6 @@
+"""Server exports for python_code_review_env."""
+
+from .app import app
+from .env import PythonCodeReviewEnvironment
+
+__all__ = ["app", "PythonCodeReviewEnvironment"]

server/app.py

@@ -0,0 +1,36 @@
+"""FastAPI entrypoint for python_code_review_env."""
+
+from __future__ import annotations
+
+try:
+    from openenv.core.env_server.http_server import create_app
+except Exception as exc:  # pragma: no cover
+    raise ImportError(
+        "openenv-core is required to run the API server. Install project dependencies first."
+    ) from exc
+
+try:
+    from ..models import PythonCodeReviewAction, PythonCodeReviewObservation
+    from .env import PythonCodeReviewEnvironment
+except ImportError:
+    from models import PythonCodeReviewAction, PythonCodeReviewObservation
+    from server.env import PythonCodeReviewEnvironment
+
+
+app = create_app(
+    PythonCodeReviewEnvironment,
+    PythonCodeReviewAction,
+    PythonCodeReviewObservation,
+    env_name="python_code_review_env",
+    max_concurrent_envs=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__":
+    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

@@ -0,0 +1,3 @@
+"""Backward-compatible import shim for the environment class."""
+
+from .env import PythonCodeReviewEnvironment, PythonCodeReviewEnvironment as PythonEnvironment

server/requirements.txt

@@ -0,0 +1,5 @@
+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)]