APPS Coding Evaluation

{
  "id": 1,
  "question": "Mikhail walks on a Cartesian plane. He starts at the point $(0, 0)$, and in one move he can go to any of eight adjacent points. For example, if Mikhail is currently at the point $(0, 0)$, he can go to any of the following points in one move: $(1, 0)$; $(1, 1)$; $(0, 1)$; $(-1, 1)$; $(-1, 0)$; $(-1, -1)$; $(0, -1)$; $(1, -1)$.\n\nIf Mikhail goes from the point $(x1, y1)$ to the point $(x2, y2)$ in one move, and $x1 \ne x2$ and $y1 \ne y2$, then such a move is called a diagonal move.\n\nMikhail has $q$ queries. For the $i$-th query Mikhail's target is to go to the point $(n_i, m_i)$ from the point $(0, 0)$ in exactly $k_i$ moves...",
  "solutions": [
    "q=int(input())\n\nfor e in range(q):\n    x,y,k=list(map(int,input().split()))\n    x,y=abs(x),abs(y)\n    x,y=max(x,y),min(x,y)\n    # ... complete solution"
  ],
  "input_output": {
    "inputs": [
      "3
      2 2 3
      4 3 7
      10 1 9"
    ],
    "outputs": [
      "1
      6
      -1"
    ]
  },
  "difficulty": "interview",
  "url": "https://codeforces.com/problemset/problem/1036/B",
  "starter_code": ""
}

import json
from typing import Any, Dict, List

from eval_protocol.models import EvaluateResult, EvaluationRow, Message
from eval_protocol.pytest import SingleTurnRolloutProcessor, evaluation_test
from eval_protocol.rewards.apps_coding_reward import evaluate_apps_solution

def apps_dataset_to_evaluation_row(data: List[Dict[str, Any]]) -> List[EvaluationRow]:
    """
    Convert entries from APPS dataset to EvaluationRow objects.
    
    This adapter extracts the problem statement and stores the comprehensive
    test cases (input/output pairs) as ground truth for evaluation.
    
    Args:
        data: List of APPS dataset entries with problem descriptions and test cases
        
    Returns:
        List of EvaluationRow objects ready for evaluation
    """
    return [
        EvaluationRow(
            messages=[Message(role="user", content=row["question"])],
            ground_truth=row["input_output"]
        )
        for row in data
    ]

@evaluation_test(
    input_dataset=["tests/pytest/data/apps_sample_dataset.jsonl"],
    dataset_adapter=apps_dataset_to_evaluation_row,
    completion_params=[{"model": "accounts/fireworks/models/kimi-k2-instruct", "temperature": 0.0, "max_tokens": 4096}],
    passed_threshold=0.33,
    rollout_processor=SingleTurnRolloutProcessor(),
    num_runs=1,
    mode="pointwise",
)
def test_apps_code_evaluation(row: EvaluationRow) -> EvaluationRow:
    """
    Evaluation function that tests APPS coding problems using evaluate_apps_solution.

    Args:
        row: EvaluationRow containing the conversation messages and ground_truth as JSON string
    
    Returns:
        EvaluationRow with the evaluation result
    """
    # Use evaluate_apps_solution directly
    result = evaluate_apps_solution(
        messages=row.messages,
        ground_truth=row.ground_truth,
    )
    
    # Set the evaluation result on the row
    row.evaluation_result = result
    
    return row

def evaluate_apps_solution(
    messages: List[Message], 
    ground_truth: Optional[str], 
    **kwargs
) -> EvaluateResult:

# Extract Python code from model response
code_solution = _extract_python_code(raw_solution_content)

# Handles various response formats:
# - Markdown code blocks: ```python ... ```
# - Inline code snippets  
# - Mixed text and code responses
# - Removes verbose explanations and comments

# Parse JSON test case data
if isinstance(ground_truth, str):
    in_outs = json.loads(ground_truth)  # Parse JSON string
elif isinstance(ground_truth, dict):
    in_outs = ground_truth  # Already parsed by JSONL loader

# Validate required structure
assert "inputs" in in_outs and "outputs" in in_outs

# Force standard input execution path and prepare secure environment
in_outs_for_check = in_outs.copy()
if "fn_name" in in_outs_for_check:
    del in_outs_for_check["fn_name"]  # Use stdin/stdout testing

# For each test case in the problem:
for i, (test_input, expected_output) in enumerate(zip(inputs, outputs)):
    # Prepare secure execution environment
    wrapped_code = f"""
import sys
sys.setrecursionlimit(6*10**5)
{standard_imports}  # Common competitive programming imports

{user_generated_code}
"""
    
    # Execute in isolated subprocess with resource limits
    process = subprocess.run(
        [sys.executable, "-c", wrapped_code],
        input=test_input,
        capture_output=True,
        timeout=timeout,
        text=True
    )
    
    # Compare outputs and record result
    if process.returncode == 0:
        actual_output = process.stdout.strip()
        results.append(actual_output == expected_output.strip())
    else:
        results.append(False)  # Runtime error

# Calculate pass rate from results
actual_results = results_list  # List of True/False for each test case
num_tests = len(actual_results)
passed_count = sum(1 for res in actual_results if res is True)
score = float(passed_count) / num_tests

# Process execution metadata for detailed error reporting
if exec_metadata_list:
    if len(exec_metadata_list) == 1 and exec_metadata_list[0].get("error"):
        # Global compilation error
        reason_msg += f" Execution Error: {exec_metadata_list[0]['error']}"
    elif score == 0.0 and exec_metadata_list[0].get("error_message") == "Wrong Answer":
        # Detailed failure analysis with specific test case details
        first_fail_meta = exec_metadata_list[0]
        reason_msg += (
            f". First fail details: Inputs: {first_fail_meta.get('inputs', 'N/A')}, "
            f"Expected: {first_fail_meta.get('expected', 'N/A')}, "
            f"Got: {first_fail_meta.get('output', 'N/A')}"
        )

# Problem: Mikhail's diagonal moves (from example above)
# Model generates solution
result = evaluate_apps_solution(
    messages=[Message(role="user", content=problem_description)],
    ground_truth='{"inputs": ["3\\n2 2 3\\n4 3 7\\n10 1 9\\n"], "outputs": ["1\\n6\\n-1\\n"]}'
)

# Result might be:
# EvaluateResult(
#     score=1.0,  # All test cases passed
#     reason="Passed 1/1 test cases",
#     metrics={
#         "pass_rate": MetricResult(score=1.0, reason="1/1"),
#         "execution_metadata": MetricResult(...)
#     }
# )

# Problem: Mikhail's diagonal moves
# Model provides optimal solution using coordinate geometry
q = int(input())
for _ in range(q):
    x, y, k = list(map(int, input().split()))
    x, y = abs(x), abs(y)
    # ... correct algorithm implementation
    # Handles all coordinate movement constraints

# Result: ✅ Passed 3/3 test cases (100% success rate)

# Problem: Mikhail's diagonal moves  
# Model has correct main logic but misses boundary condition
q = int(input())
for _ in range(q):
    x, y, k = list(map(int, input().split()))
    # ... mostly correct implementation
    # Fails on impossible movement case

# Result: ⚠️ Passed 2/3 test cases (67% success rate)

# Problem: Mikhail's diagonal moves
# Model uses incorrect movement calculation
q = int(input())
for _ in range(q):
    x, y, k = list(map(int, input().split()))
    # Incorrect approach - doesn't consider diagonal optimization
    print(k)  # Always outputs k regardless of constraints

# Result: ❌ Passed 0/3 test cases - Wrong algorithmic approach

# Problem: Mikhail's diagonal moves
# Model uses inefficient brute force instead of mathematical approach
q = int(input())
for _ in range(q):
    x, y, k = list(map(int, input().split()))
    # Simulates all possible paths - exponential complexity
    # Times out on larger coordinate values

# Result: ❌ Execution timeout - Algorithm too slow for constraints

# Problem: Mikhail's diagonal moves
# Model has syntax errors
q = int(input())
for _ in range(q)  # Missing colon
    x, y, k = list(map(int, input().split()))
    # ... rest of solution

# Result: ❌ Compilation error: SyntaxError - Invalid Python syntax