src/eval/code_evaluator.py - master - Watashicuvu/agentic-tools

src/eval/code_evaluator.py
360 строк11 KB
Якуб
mts update
12 апр 2026, 21:19
12 апр 2026, 21:19bff6dcd
"""Abstract CodeEvaluator base class + language-specific implementations.

Provides a unified interface for code evaluation across languages:
- syntax_check(code) → bool
- lint(code, work_dir) → LintResult
- run(code, work_dir, timeout) → RunResult

Language-specific implementations inherit from CodeEvaluator and override
the abstract methods.
"""

from __future__ import annotations

import asyncio
import shlex
import tempfile
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from pathlib import Path
from typing import Optional, Tuple

from src.services.telemetry import telemetry, EventType


@dataclass(frozen=True)
class LintResult:
    """Result of a linting check.

    Attributes:
        ok: Whether the check passed.
        stdout: Standard output from the linter.
        stderr: Standard error output from the linter.
        errors: Parsed list of error messages (empty if ok).
    """

    ok: bool
    stdout: str = ""
    stderr: str = ""
    errors: list[str] = field(default_factory=list)


@dataclass(frozen=True)
class RunResult:
    """Result of code execution.

    Attributes:
        ok: Whether execution succeeded.
        output: Combined stdout/stderr.
        error: Error message if execution failed (None if ok).
    """

    ok: bool
    output: str = ""
    error: Optional[str] = None


@dataclass(frozen=True)
class CodeEvalResult:
    """Complete result of code evaluation.

    Attributes:
        code_syntax_ok: Syntax check passed.
        code_syntax_output: Syntax checker output.
        code_lint_ok: Lint check passed.
        code_lint_errors: List of lint error messages.
        code_lint_output: Full linter output.
        code_run_ok: Execution passed.
        code_run_output: Program stdout/stderr.
        code_run_error: Error message if execution failed.
        execution_time_ms: Time spent on evaluation.
    """

    code_syntax_ok: bool = False
    code_syntax_output: str = ""
    code_lint_ok: bool = False
    code_lint_errors: list[str] = field(default_factory=list)
    code_lint_output: str = ""
    code_run_ok: bool = False
    code_run_output: str = ""
    code_run_error: Optional[str] = None
    execution_time_ms: float = 0.0


# Default timeout for code execution (seconds)
_DEFAULT_TIMEOUT = 30


class CodeEvaluator(ABC):
    """Abstract base class for language-specific code evaluators.

    Subclasses must implement:
    - _check_syntax(code, work_dir) → LintResult
    - _run_linter(code, work_dir) → LintResult
    - _run_code(code, work_dir) → RunResult
    - _language_name() → str
    - _syntax_command(file_path) → list[str]
    - _lint_command(file_path) → list[str]
    - _run_command(file_path) → list[str]

    Args:
        timeout: Maximum seconds for each subprocess call.
    """

    def __init__(self, timeout: int = _DEFAULT_TIMEOUT) -> None:
        self.timeout = timeout

    @abstractmethod
    def _language_name(self) -> str:
        """Return human-readable language name."""
        ...

    @abstractmethod
    def _syntax_command(self, file_path: str) -> list[str]:
        """Return command for syntax check."""
        ...

    @abstractmethod
    def _lint_command(self, file_path: str) -> list[str]:
        """Return command for linting."""
        ...

    @abstractmethod
    def _run_command(self, file_path: str) -> list[str]:
        """Return command for execution."""
        ...

    @abstractmethod
    def _parse_lint_errors(self, returncode: int, stdout: str, stderr: str) -> list[str]:
        """Parse lint errors from command output."""
        ...

    async def evaluate(
        self,
        code: str,
        work_dir: Optional[str] = None,
        run_code: bool = True,
    ) -> CodeEvalResult:
        """Run full evaluation pipeline.

        Steps:
        1. Syntax check
        2. Linting — if syntax passed
        3. Execution — if run_code=True and lint passed

        Args:
            code: Source code to evaluate.
            work_dir: Working directory for execution.
            run_code: Whether to execute the code.

        Returns:
            CodeEvalResult with all validation results.
        """
        import time

        start = time.monotonic()
        work_path = Path(work_dir) if work_dir else None
        lang = self._language_name()

        # Step 1: Syntax check
        syntax_result = await self._check_syntax(code, work_path)
        if not syntax_result.ok:
            elapsed = (time.monotonic() - start) * 1000
            telemetry.emit(
                EventType.STEP_INFO,
                f"{lang} syntax check failed",
                {"output": syntax_result.stderr[:200]},
            )
            return CodeEvalResult(
                code_syntax_ok=False,
                code_syntax_output=syntax_result.stderr,
                execution_time_ms=round(elapsed, 2),
            )

        # Step 2: Linting
        lint_result = await self._run_linter(code, work_path)
        if not lint_result.ok:
            elapsed = (time.monotonic() - start) * 1000
            telemetry.emit(
                EventType.STEP_INFO,
                f"{lang} lint check failed",
                {"errors": lint_result.errors[:5]},
            )
            return CodeEvalResult(
                code_syntax_ok=True,
                code_lint_ok=False,
                code_lint_errors=lint_result.errors,
                code_lint_output=lint_result.stdout + lint_result.stderr,
                execution_time_ms=round(elapsed, 2),
            )

        # Step 3: Execution (optional)
        run_result: RunResult = RunResult(ok=True)
        if run_code:
            run_result = await self._run_code(code, work_path)

        elapsed = (time.monotonic() - start) * 1000

        telemetry.emit(
            EventType.STEP_INFO,
            f"{lang} evaluation complete",
            {
                "code_syntax_ok": True,
                "code_lint_ok": True,
                "code_run_ok": run_result.ok,
                "time_ms": round(elapsed, 2),
            },
        )

        return CodeEvalResult(
            code_syntax_ok=True,
            code_lint_ok=True,
            code_run_ok=run_result.ok,
            code_run_output=run_result.output,
            code_run_error=run_result.error,
            execution_time_ms=round(elapsed, 2),
        )

    async def _check_syntax(
        self,
        code: str,
        work_dir: Optional[Path],
    ) -> LintResult:
        """Check syntax using language-specific tool."""
        temp_file = await self._write_temp_file(code, "temp_syntax", work_dir)
        try:
            cmd = self._syntax_command(str(temp_file))
            returncode, stdout, stderr = await self._run_subprocess(cmd)

            return LintResult(
                ok=returncode == 0,
                stdout=stdout,
                stderr=stderr,
                errors=[stderr.strip()] if returncode != 0 else [],
            )
        finally:
            await self._cleanup_temp_file(temp_file)

    async def _run_linter(
        self,
        code: str,
        work_dir: Optional[Path],
    ) -> LintResult:
        """Run linter static analysis."""
        temp_file = await self._write_temp_file(code, "temp_lint", work_dir)
        try:
            cmd = self._lint_command(str(temp_file))
            returncode, stdout, stderr = await self._run_subprocess(cmd)
            errors = self._parse_lint_errors(returncode, stdout, stderr)

            return LintResult(
                ok=returncode == 0,
                stdout=stdout,
                stderr=stderr,
                errors=errors,
            )
        finally:
            await self._cleanup_temp_file(temp_file)

    async def _run_code(
        self,
        code: str,
        work_dir: Optional[Path],
    ) -> RunResult:
        """Execute code and capture output."""
        temp_file = await self._write_temp_file(code, "temp_run", work_dir)
        try:
            cmd = self._run_command(str(temp_file))
            returncode, stdout, stderr = await self._run_subprocess(cmd)

            output = stdout + stderr
            if returncode == 0:
                return RunResult(ok=True, output=output)
            else:
                return RunResult(
                    ok=False,
                    output=output,
                    error=stderr.strip() if stderr else output.strip(),
                )
        except asyncio.TimeoutError:
            return RunResult(
                ok=False,
                error=f"Execution timed out after {self.timeout}s",
            )
        finally:
            await self._cleanup_temp_file(temp_file)

    async def _run_subprocess(
        self,
        cmd: list[str],
    ) -> tuple[int, str, str]:
        """Run a shell command with timeout.

        Args:
            cmd: Command and arguments as list.

        Returns:
            Tuple of (returncode, stdout, stderr).

        Raises:
            asyncio.TimeoutError: If command exceeds timeout.
        """
        telemetry.emit(
            EventType.STEP_INFO,
            "Running command",
            {"cmd": cmd, "timeout": self.timeout},
        )

        try:
            proc = await asyncio.create_subprocess_exec(
                *cmd,
                stdout=asyncio.subprocess.PIPE,
                stderr=asyncio.subprocess.PIPE,
            )
            stdout_bytes, stderr_bytes = await asyncio.wait_for(
                proc.communicate(), timeout=self.timeout
            )
            stdout = stdout_bytes.decode("utf-8", errors="replace")
            stderr = stderr_bytes.decode("utf-8", errors="replace")
            return proc.returncode or 0, stdout, stderr
        except asyncio.TimeoutError:
            proc.kill()
            telemetry.emit(
                EventType.ERROR,
                "Command timed out",
                {"cmd": cmd, "timeout": self.timeout},
            )
            raise

    async def _write_temp_file(
        self,
        code: str,
        filename_base: str,
        work_dir: Optional[Path],
    ) -> Path:
        """Write code to a temporary file with language-specific extension."""
        ext = self._file_extension()
        filename = f"{filename_base}{ext}"

        if work_dir:
            temp_path = work_dir / filename
        else:
            temp_dir = Path(tempfile.gettempdir())
            temp_path = temp_dir / filename

        temp_path.write_text(code, encoding="utf-8")
        return temp_path

    @abstractmethod
    def _file_extension(self) -> str:
        """Return file extension (e.g., '.py', '.lua')."""
        ...

    @staticmethod
    async def _cleanup_temp_file(path: Path) -> None:
        """Remove temporary file."""
        try:
            if path.exists():
                path.unlink()
        except OSError:
            pass  # Best-effort cleanup