Deploy: server/grader.py

server/grader.py

@@ -0,0 +1,128 @@
+from .models import ExecutionResult, TaskInfo
+from .llm_judge import llm_judge
+
+def force_valid_reward(value) -> float:
+    """Hard guarantee: reward is strictly in (0, 1) — never 0 or 1, no exceptions."""
+    try:
+        r = float(value)
+    except Exception:
+        return 0.5
+
+    if r <= 0.001:
+        return 0.001
+    if r >= 0.999:
+        return 0.999
+
+    return r
+
+def safe_reward(reward) -> float:
+    """Clamp reward to open interval (0, 1) via force_valid_reward."""
+    if reward is None:
+        reward = 0.5
+    return force_valid_reward(reward)
+
+def normalize_reward(passed: int, total: int) -> float:
+    if total == 0:
+        return 0.5
+    raw = passed / total
+    return force_valid_reward(raw)
+
+def get_llm_quality_score(proposed_fix: str) -> dict:
+    return llm_judge("", proposed_fix, "unknown")
+
+
+def calculate_codearena_reward(
+    *,
+    compile_ok: bool,
+    passed: int,
+    total: int,
+    execution_time_seconds: float,
+    optimal_time_seconds: float,
+    buggy_code: str,
+    proposed_fix: str,
+    task_category: str,
+    step_count: int,
+    is_repeated_fix: bool,
+) -> tuple[float, dict]:
+    compile_score = 1.0 if compile_ok else 0.0
+    test_pass_ratio = passed / total if total else 0.0
+
+    efficiency_score = 0.0
+    if test_pass_ratio == 1.0:
+        if execution_time_seconds <= optimal_time_seconds:
+            efficiency_score = 1.0
+        else:
+            ratio = execution_time_seconds / max(0.001, optimal_time_seconds)
+            efficiency_score = max(0.0, 1.0 - (ratio - 1.0) / 2.0)
+
+    llm_scores = llm_judge(buggy_code, proposed_fix, task_category)
+    llm_correctness = float(llm_scores.get("correctness", 0.5))
+    llm_security = float(llm_scores.get("security", 0.5))
+    llm_code_quality = float(llm_scores.get("code_quality", 0.5))
+    llm_judge_score = (llm_correctness + llm_security + llm_code_quality) / 3
+    novelty_penalty = 1.0 if is_repeated_fix else 0.0
+    step_penalty = 0.02 * step_count
+
+    final_reward = (
+        0.20 * compile_score
+        + 0.40 * test_pass_ratio
+        + 0.10 * efficiency_score
+        + 0.30 * llm_judge_score
+        - step_penalty
+        - 0.10 * novelty_penalty
+    )
+    final_reward = force_valid_reward(round(final_reward, 4))
+
+    return final_reward, {
+        "compile_score": compile_score,
+        "test_pass_ratio": test_pass_ratio,
+        "efficiency_score": efficiency_score,
+        "llm_correctness": llm_correctness,
+        "llm_security": llm_security,
+        "llm_code_quality": llm_code_quality,
+        "step_penalty": step_penalty,
+        "novelty_penalty": novelty_penalty,
+    }
+
+def calculate_reward_components(exec_result: ExecutionResult, task_info: TaskInfo, proposed_fix: str) -> dict:
+    _, components = calculate_codearena_reward(
+        compile_ok=exec_result.compile_success,
+        passed=exec_result.test_passed,
+        total=exec_result.test_total,
+        execution_time_seconds=exec_result.execution_time_seconds,
+        optimal_time_seconds=task_info.optimal_time_seconds,
+        buggy_code=task_info.buggy_code,
+        proposed_fix=proposed_fix,
+        task_category=task_info.difficulty,
+        step_count=0,
+        is_repeated_fix=False,
+    )
+    return components
+
+def calculate_reward(
+    exec_result: ExecutionResult,
+    task_info: TaskInfo,
+    proposed_fix: str,
+    step_count: int = 0,
+    is_repeated_fix: bool = False,
+) -> tuple[float, dict]:
+    return calculate_codearena_reward(
+        compile_ok=exec_result.compile_success,
+        passed=exec_result.test_passed,
+        total=exec_result.test_total,
+        execution_time_seconds=exec_result.execution_time_seconds,
+        optimal_time_seconds=task_info.optimal_time_seconds,
+        buggy_code=task_info.buggy_code,
+        proposed_fix=proposed_fix,
+        task_category=task_info.difficulty,
+        step_count=step_count,
+        is_repeated_fix=is_repeated_fix,
+    )
+
+def grade(*args, **kwargs) -> float:
+    try:
+        if len(args) == 3:
+            return calculate_reward(args[0], args[1], args[2])[0]
+        return 0.5
+    except Exception:
+        return 0.5