ariadne/ariadne/services/hermes_code_defects.py

"""Recognise mechanically-fixable defects in a build's console evidence.

The bounded patcher works best when it is told precisely what to fix rather
than left to infer it. A linter has already done that work: it names the file,
the line, the rule, and what is wrong. Extracting that turns an open-ended
"repair this build" request into a narrow instruction whose result can be
checked against the same evidence.

Only categories whose fix is mechanical belong here. A defect that needs a
judgement call about intended behaviour is not one of them; that stays with
the ordinary escalation path.
"""

from __future__ import annotations

import re
from typing import Any


LINT_VIOLATION = "lint_violation"

# Each pattern must capture path, line, rule and message. Anything that cannot
# name all four is not actionable enough to narrow the patch instruction, so it
# is deliberately absent rather than partially matched.
_PATTERNS: tuple[tuple[str, re.Pattern[str]], ...] = (
    # ruff / flake8:  path.py:12:5: E501 Line too long (95 > 88)
    (
        "ruff",
        re.compile(
            r"^(?P<path>[\w./\-]+\.py):(?P<line>\d+):\d+:\s+(?P<rule>[A-Z]+\d+)\s+(?P<message>.+)$"
        ),
    ),
    # eslint (stylish):  12:5  error  Unexpected console statement  no-console
    (
        "eslint",
        re.compile(
            r"^\s*(?P<line>\d+):\d+\s+(?:error|warning)\s+(?P<message>.+?)\s\s+(?P<rule>[\w\-/]+)$"
        ),
    ),
    # golangci-lint:  path.go:12:5: msg (govet)
    (
        "golangci",
        re.compile(
            r"^(?P<path>[\w./\-]+\.go):(?P<line>\d+):\d+:\s+(?P<message>.+?)\s+\((?P<rule>[\w\-]+)\)$"
        ),
    ),
)

_ESLINT_FILE = re.compile(r"^(?P<path>[\w./\-]+\.(?:js|jsx|ts|tsx))$")
_MAX_DEFECTS = 20


def extract_lint_defects(bundle: dict, cfg: dict) -> list[dict[str, Any]]:
    """Return the linter diagnostics a build's console evidence names.

    Inputs: the evidence bundle from `collect_evidence` and the per-job code
    cfg, whose `allowed_path_prefixes` and `allowed_suffixes` bound which files
    may be reported. Outputs: at most twenty
    {"category", "tool", "path", "line", "rule", "message"} dicts in the order
    encountered, deduplicated by path/line/rule.

    A diagnostic outside the writable allowlist is dropped here rather than
    later, so a defect is never reported for a file that could not be patched
    anyway. Never raises; returns [] when nothing matches.
    """

    try:
        return _scan(_evidence_lines(bundle), cfg)
    except Exception:
        return []


def _evidence_lines(bundle: dict) -> list[str]:
    """Flatten the console failure regions and tail into scannable lines."""

    jenkins = bundle.get("jenkins") if isinstance(bundle.get("jenkins"), dict) else {}
    text: list[str] = []
    regions = jenkins.get("console_failures")
    for region in regions if isinstance(regions, list) else []:
        if isinstance(region, dict):
            text.extend(str(region.get("text") or "").split("\n"))
    text.extend(str(jenkins.get("console_tail") or "").split("\n"))
    return text


def _scan(lines: list[str], cfg: dict) -> list[dict[str, Any]]:
    """Match every known diagnostic shape over the evidence lines."""

    found: list[dict[str, Any]] = []
    seen: set[tuple[str, str, str]] = set()
    eslint_path = ""
    for raw in lines:
        line = raw.rstrip()
        file_only = _ESLINT_FILE.match(line.strip())
        if file_only:
            eslint_path = file_only.group("path")
            continue
        for tool, pattern in _PATTERNS:
            match = pattern.match(line)
            if not match:
                continue
            path = match.groupdict().get("path") or eslint_path
            if not path or not _is_writable(path, cfg):
                break
            key = (path, match.group("line"), match.group("rule"))
            if key in seen:
                break
            seen.add(key)
            found.append(
                {
                    "category": LINT_VIOLATION,
                    "tool": tool,
                    "path": path,
                    "line": int(match.group("line")),
                    "rule": match.group("rule"),
                    "message": match.group("message").strip(),
                }
            )
            break
        if len(found) >= _MAX_DEFECTS:
            break
    return found


def _is_writable(path: str, cfg: dict) -> bool:
    """Report whether a diagnostic's file is one the patcher may write to."""

    prefixes = [str(p) for p in (cfg.get("allowed_path_prefixes") or [])]
    suffixes = [str(s) for s in (cfg.get("allowed_suffixes") or [])]
    if suffixes and not any(path.endswith(s) for s in suffixes):
        return False
    return bool(prefixes) and any(path.startswith(p) for p in prefixes)


def defect_instruction(defects: list[dict[str, Any]]) -> str:
    """Render the defects as a precise instruction for the patch prompt.

    Inputs: the defects from `extract_lint_defects`. Outputs: a prompt
    fragment naming each one, or "" when there are none so the caller falls
    back to the open-ended request. Naming the rule and line is the point:
    it converts "repair this build" into a change whose correctness can be
    read off the same evidence.
    """

    if not defects:
        return ""
    lines = [
        "A linter reported these violations; fix exactly these and nothing else:",
    ]
    for defect in defects:
        lines.append(
            f"- {defect['path']} line {defect['line']}: {defect['rule']} {defect['message']}"
        )
    lines.append("Do not reformat unrelated lines and do not suppress the rule.")
    return "\n".join(lines)