orchestrata_ultimate.harness.evaluators.lint

Reference › orchestrata_ultimate › harness › evaluators › lint

LintEvaluator, registered as lint_check (SPEC-102 US-005, REQ-102-065). It runs ruff check on each resolved file as a subprocess of the form interpreter, -m ruff, so the tool comes from the interpreter's own module path rather than from whatever PATH holds (ADR-0089); the interpreter defaults to the running one, and an environment mapping can be passed. Output from a non-zero exit becomes an issue; a non-zero exit with no output also counts when the severity param is "ERROR" or "WARNING". Two infrastructure cases are reported as failures with their own message, distinct from lint findings: ruff not installed for the interpreter, and the interpreter not found. Any issue gives score 0.0, otherwise 1.0. A comparable commit-time check is the ruff gate. See the evaluator registry.

At a glance

Class diagram

Dependency diagram

API

Classes

LintEvaluator

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

Run ruff check on generated/modified files.

Source

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

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

from __future__ import annotations

import asyncio
import sys
from collections.abc import Mapping
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-065"])
class LintEvaluator(Evaluator):
    """Run ``ruff check`` on generated/modified files.

    The *severity* config parameter controls the strictness:

    - ``"ERROR"`` (default): any finding from ``ruff check`` is treated as a
      failure.
    - ``"WARNING"``: same behaviour — ``ruff check`` reports all enabled rules
      and does not natively distinguish severity levels in its default output.
      Both modes fail when ruff reports any issue.
    """

    name = "lint_check"

    @traces(spec="102", story="US-005", reqs=["REQ-102-065"])
    async def evaluate(
        self,
        output: Any,
        context: HarnessContext,
        config: EvaluatorConfig,
        env: Mapping[str, str] | None = None,
        interpreter: str | None = None,
    ) -> EvaluatorResult:
        """Run ``ruff check`` on the specified files.

        Args:
            output: file(s) to lint, resolved by :meth:`_resolve_files`.
            context: harness context supplying the worktree root.
            config: evaluator config; ``severity`` labels the failure detail.
            env: environment for the spawned subprocess (``None`` inherits
                this process's). It no longer decides WHERE ruff comes from —
                the tool is resolved through the interpreter's module path.
            interpreter: the Python that runs ruff as ``-m ruff``. Defaults to
                ``sys.executable``, the running interpreter, per ADR-0089: a
                Python tool is spawned as ``[<interpreter>, "-m", <tool>]``,
                never as a bare name resolved by whatever PATH the session
                happens to carry.
        """
        if interpreter is None:
            interpreter = sys.executable
        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.",
            )

        severity = config.params.get("severity", "ERROR")
        all_issues: list[str] = []
        checked = 0

        for file_path in files:
            if not file_path.exists():
                all_issues.append(f"{file_path}: file not found")
                continue
            checked += 1
            try:
                proc = await asyncio.create_subprocess_exec(
                    interpreter,
                    "-m",
                    "ruff",
                    "check",
                    str(file_path),
                    stdout=asyncio.subprocess.PIPE,
                    stderr=asyncio.subprocess.PIPE,
                    env=env,
                )
                stdout_bytes, stderr_bytes = await proc.communicate()
            except FileNotFoundError:
                return EvaluatorResult(
                    name=self.name,
                    passed=False,
                    score=0.0,
                    details=f"Python interpreter not found: {interpreter}",
                )

            if proc.returncode != 0:
                stdout = stdout_bytes.decode("utf-8", errors="replace").strip()
                stderr = stderr_bytes.decode("utf-8", errors="replace").strip()
                # BEFORE the "non-zero exit with output = lint findings"
                # branch: a missing ruff MODULE is an infrastructure verdict
                # about the interpreter, and folding its traceback in with
                # lint findings would report "the code is dirty" when the
                # truth is "the tool never ran".
                if "No module named ruff" in stderr:
                    return EvaluatorResult(
                        name=self.name,
                        passed=False,
                        score=0.0,
                        details=(
                            f"ruff is not installed for {interpreter} "
                            "(python -m ruff: No module named ruff)"
                        ),
                    )
                output_text = "\n".join(filter(None, [stdout, stderr]))
                if output_text:
                    all_issues.append(f"{file_path}:\n{output_text}")
                elif severity in ("ERROR", "WARNING"):
                    # ruff exited non-zero but produced no output — treat as issue
                    all_issues.append(
                        f"{file_path}: ruff check failed (exit code {proc.returncode})"
                    )

        if all_issues:
            return EvaluatorResult(
                name=self.name,
                passed=False,
                score=0.0,
                details=f"Severity: {severity}\n\n" + "\n\n".join(all_issues),
            )

        return EvaluatorResult(
            name=self.name,
            passed=True,
            score=1.0,
            details=f"All {checked} file(s) passed ruff check (severity={severity}).",
        )

    @traces(spec="102", story="US-005", reqs=["REQ-102-065"])
    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