feat(graders): add ASTCodeGrader and is_valid_python static code eval…

hud/graders/__init__.py

@@ -25,6 +25,7 @@
 
 from .base import Grader
 from .bash import BashGrader
+from .code import ASTCodeGrader, is_valid_python
 from .combine import _combine_subscores, combine, combine_all, combine_any
 from .judge import LLMJudgeGrader
 from .results import EvaluationResult, SubScore
@@ -39,6 +40,7 @@
 )
 
 __all__ = [
+    "ASTCodeGrader",
     "BashGrader",
     "EvaluationResult",
     "Grader",
@@ -53,6 +55,7 @@
     "contains_any",
     "exact_match",
     "f1_score",
+    "is_valid_python",
     "normalize",
     "numeric_match",
 ]

hud/graders/code.py

@@ -0,0 +1,139 @@
+"""``ASTCodeGrader`` — evaluate Python code AST and enforce anti-cheat policies."""
+
+from __future__ import annotations
+
+import ast
+import logging
+from typing import Any
+
+from .base import Grader
+from .results import SubScore
+
+logger = logging.getLogger(__name__)
+
+
+class ASTCodeGrader(Grader):
+    """Static Python AST evaluator for code generation rollouts and anti-cheat policies.
+
+    Validates Python syntax without executing untrusted code, and optionally verifies
+    required function/class definitions while enforcing anti-cheat rules (e.g. disallowed
+    imports such as 'subprocess', 'os', 'sys' or forbidden calls such as 'eval', 'exec').
+    """
+
+    name = "ASTCodeGrader"
+
+    @classmethod
+    async def compute_score(
+        cls,
+        code: str | None = None,
+        required_functions: list[str] | None = None,
+        required_classes: list[str] | None = None,
+        disallowed_imports: list[str] | None = None,
+        disallowed_calls: list[str] | None = None,
+        **kwargs: Any,
+    ) -> SubScore:
+        """Parse ``code`` into an AST and enforce syntactic/anti-cheat constraints."""
+        if code is None:
+            raise ValueError("ASTCodeGrader requires code")
+        del kwargs
+
+        # 1. Syntax parse
+        try:
+            tree = ast.parse(code)
+        except SyntaxError as err:
+            logger.debug("ASTCodeGrader syntax error on line %s: %s", err.lineno, err.msg)
+            return SubScore(
+                name=cls.name,
+                value=0.0,
+                info={
+                    "valid_syntax": False,
+                    "error": str(err.msg),
+                    "lineno": err.lineno,
+                    "offset": err.offset,
+                },
+            )
+
+        # 2. Extract defined functions, classes, imports, and calls
+        defined_funcs = {
+            node.name
+            for node in ast.walk(tree)
+            if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef))
+        }
+        defined_classes = {
+            node.name for node in ast.walk(tree) if isinstance(node, ast.ClassDef)
+        }
+
+        imports: set[str] = set()
+        for node in ast.walk(tree):
+            if isinstance(node, ast.Import):
+                for alias in node.names:
+                    imports.add(alias.name.split(".")[0])
+            elif isinstance(node, ast.ImportFrom) and node.module:
+                imports.add(node.module.split(".")[0])
+
+        calls: set[str] = set()
+        for node in ast.walk(tree):
+            if isinstance(node, ast.Call):
+                if isinstance(node.func, ast.Name):
+                    calls.add(node.func.id)
+                elif isinstance(node.func, ast.Attribute):
+                    calls.add(node.func.attr)
+
+        violations: list[str] = []
+        missing_definitions: list[str] = []
+
+        if required_functions:
+            missing_definitions.extend(
+                f"Missing required function: {fn}"
+                for fn in required_functions
+                if fn not in defined_funcs
+            )
+
+        if required_classes:
+            missing_definitions.extend(
+                f"Missing required class: {cls_name}"
+                for cls_name in required_classes
+                if cls_name not in defined_classes
+            )
+
+        if disallowed_imports:
+            violations.extend(
+                f"Disallowed import detected: {imp}"
+                for imp in disallowed_imports
+                if imp in imports
+            )
+
+        if disallowed_calls:
+            violations.extend(
+                f"Disallowed call detected: {call_name}"
+                for call_name in disallowed_calls
+                if call_name in calls
+            )
+
+        passed = len(violations) == 0 and len(missing_definitions) == 0
+
+        return SubScore(
+            name=cls.name,
+            value=1.0 if passed else 0.0,
+            info={
+                "valid_syntax": True,
+                "passed": passed,
+                "defined_functions": sorted(defined_funcs),
+                "defined_classes": sorted(defined_classes),
+                "imports": sorted(imports),
+                "violations": violations,
+                "missing_definitions": missing_definitions,
+            },
+        )
+
+
+def is_valid_python(code: str) -> float:
+    """Return 1.0 if ``code`` parses as valid Python syntax without execution, else 0.0."""
+    try:
+        ast.parse(code)
+        return 1.0
+    except SyntaxError:
+        return 0.0
+
+
+__all__ = ["ASTCodeGrader", "is_valid_python"]

hud/tests/test_graders.py

@@ -12,6 +12,7 @@
 import pytest
 
 from hud.graders import (
+    ASTCodeGrader,
     BashGrader,
     EvaluationResult,
     Grader,
@@ -25,6 +26,7 @@
     contains_any,
     exact_match,
     f1_score,
+    is_valid_python,
     normalize,
     numeric_match,
 )
@@ -646,3 +648,82 @@ async def test_grade_and_combine_compose(self) -> None:
         assert by_name["BashGrader-2"].info is not None
         assert by_name["BashGrader-2"].info["exit_code"] != 0
         assert result.info == {}
+
+
+class TestASTCodeGrader:
+    def test_is_valid_python_helper(self) -> None:
+        assert is_valid_python("def add(a, b):\n    return a + b\n") == 1.0
+        assert is_valid_python("def broken(: return") == 0.0
+
+    async def test_ast_grader_valid_code(self) -> None:
+        code = "def solve(x):\n    return x * 2\n"
+        subscore = await ASTCodeGrader.compute_score(code=code)
+        assert subscore.value == 1.0
+        assert subscore.info is not None
+        assert subscore.info["valid_syntax"] is True
+        assert subscore.info["passed"] is True
+        assert "solve" in subscore.info["defined_functions"]
+
+    async def test_ast_grader_syntax_error(self) -> None:
+        bad_code = "def incomplete("
+        subscore = await ASTCodeGrader.compute_score(code=bad_code)
+        assert subscore.value == 0.0
+        assert subscore.info is not None
+        assert subscore.info["valid_syntax"] is False
+        assert "error" in subscore.info
+
+    async def test_ast_grader_required_definitions(self) -> None:
+        code = "class AgentRunner:\n    def run(self):\n        pass\n"
+        # Passing case
+        subscore = await ASTCodeGrader.compute_score(
+            code=code,
+            required_classes=["AgentRunner"],
+            required_functions=["run"],
+        )
+        assert subscore.value == 1.0
+        assert subscore.info is not None
+        assert subscore.info["passed"] is True
+
+        # Missing function
+        subscore_missing = await ASTCodeGrader.compute_score(
+            code=code,
+            required_functions=["evaluate"],
+        )
+        assert subscore_missing.value == 0.0
+        assert subscore_missing.info is not None
+        assert any(
+            "Missing required function: evaluate" in msg
+            for msg in subscore_missing.info["missing_definitions"]
+        )
+
+    async def test_ast_grader_anti_cheat_disallowed_imports(self) -> None:
+        cheating_code = "import os\nfrom subprocess import Popen\ndef hack():\n    pass\n"
+        subscore = await ASTCodeGrader.compute_score(
+            code=cheating_code,
+            disallowed_imports=["os", "subprocess"],
+        )
+        assert subscore.value == 0.0
+        assert subscore.info is not None
+        assert subscore.info["passed"] is False
+        assert len(subscore.info["violations"]) == 2
+        assert any("os" in v for v in subscore.info["violations"])
+        assert any("subprocess" in v for v in subscore.info["violations"])
+
+    async def test_ast_grader_anti_cheat_disallowed_calls(self) -> None:
+        risky_code = "def run(x):\n    return eval(x)\n"
+        subscore = await ASTCodeGrader.compute_score(
+            code=risky_code,
+            disallowed_calls=["eval", "exec"],
+        )
+        assert subscore.value == 0.0
+        assert subscore.info is not None
+        assert any("eval" in v for v in subscore.info["violations"])
+
+    async def test_ast_grader_grade_in_combine(self) -> None:
+        code = "def valid(): return True"
+        result = await combine(
+            ASTCodeGrader.grade(weight=1.0, code=code, required_functions=["valid"])
+        )
+        assert result.reward == 1.0
+        assert result.subscores is not None
+        assert result.subscores[0].name == "ASTCodeGrader"