orchestrata_ultimate.harness.evaluators.syntax

Reference › orchestrata_ultimate › harness › evaluators › syntax

SyntaxEvaluator, registered as syntax_check (SPEC-102 US-005, REQ-102-064). It resolves the files to check from the scenario output when that contains paths, or from the files param, relative to the worktree. Each .py file is read and parsed with ast.parse in a worker thread; other files are skipped and named in the details. Any syntax error fails the evaluation with score 0.0 and the traceback in the details; no files means a pass with score 1.0. It does not execute the code. See the evaluator registry.

At a glance

Class diagram

Dependency diagram

API

Classes

SyntaxEvaluator

Bases: Evaluator · Decorators: register_evaluator, traces(spec='102', story='US-005', reqs=['REQ-102-064']) · Line 23

Evaluate Python files for syntax errors using ast.parse().

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/harness/evaluators/syntax.py

"""Syntax evaluator for the Volk Test Harness."""

from __future__ import annotations

import ast
import asyncio
import traceback
from pathlib import Path
from typing import Any

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.harness.evaluators import register_evaluator
from orchestrata_ultimate.harness.models import (
    Evaluator,
    EvaluatorConfig,
    EvaluatorResult,
    HarnessContext,
)


@register_evaluator
@traces(spec="102", story="US-005", reqs=["REQ-102-064"])
class SyntaxEvaluator(Evaluator):
    """Evaluate Python files for syntax errors using ``ast.parse()``."""

    name = "syntax_check"

    @traces(spec="102", story="US-005", reqs=["REQ-102-064"])
    async def evaluate(
        self,
        output: Any,
        context: HarnessContext,
        config: EvaluatorConfig,
    ) -> EvaluatorResult:
        """Parse each Python file and report syntax errors.

        Files are resolved from *output* (when it contains paths) or from
        ``config.params["files"]``. Non-Python files are skipped with a
        warning recorded in the result details.
        """
        files = self._resolve_files(output, config, context.worktree_path)
        if not files:
            return EvaluatorResult(
                name=self.name,
                passed=True,
                score=1.0,
                details="No files to check.",
            )

        errors: list[str] = []
        checked = 0
        skipped: list[str] = []

        for file_path in files:
            if file_path.suffix != ".py":
                skipped.append(str(file_path))
                continue
            checked += 1
            try:
                content = await asyncio.to_thread(file_path.read_text, encoding="utf-8")
                await asyncio.to_thread(ast.parse, content)
            except SyntaxError:
                errors.append(f"{file_path}:\n{traceback.format_exc().strip()}")
            except Exception as exc:  # pragma: no cover
                errors.append(f"{file_path}: {exc}")

        details_parts: list[str] = []
        if checked:
            details_parts.append(f"Checked {checked} Python file(s).")
        if skipped:
            details_parts.append(
                f"Skipped {len(skipped)} non-Python file(s): {', '.join(skipped)}"
            )
        if errors:
            details_parts.append("Errors:\n" + "\n\n".join(errors))
            return EvaluatorResult(
                name=self.name,
                passed=False,
                score=0.0,
                details="\n".join(details_parts),
            )

        return EvaluatorResult(
            name=self.name,
            passed=True,
            score=1.0,
            details=(
                "\n".join(details_parts)
                if details_parts
                else "All files parsed successfully."
            ),
        )

    @traces(spec="102", story="US-005", reqs=["REQ-102-064"])
    def _resolve_files(
        self,
        output: Any,
        config: EvaluatorConfig,
        worktree_path: Path,
    ) -> list[Path]:
        """Resolve file paths from *output* or config params."""
        raw_paths: list[str] = []

        if isinstance(output, (str, Path)):
            raw_paths = [str(output)]
        elif isinstance(output, (list, tuple)):
            raw_paths = [str(p) for p in output]
        elif isinstance(output, dict):
            for key in ("files", "file_paths", "paths"):
                if key in output:
                    val = output[key]
                    if isinstance(val, (list, tuple)):
                        raw_paths = [str(p) for p in val]
                    else:
                        raw_paths = [str(val)]
                    break

        if not raw_paths:
            raw_paths = config.params.get("files", [])

        resolved: list[Path] = []
        for p in raw_paths:
            path = Path(p)
            if not path.is_absolute():
                path = worktree_path / path
            resolved.append(path)
        return resolved