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"
UNDEFINED_NAME = "undefined_name"
FAILING_ASSERTION = "failing_assertion"
SONARQUBE_ISSUE = "sonarqube_issue"

# 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))$")

# A missing or misspelled name is named exactly by the runtime or compiler, so
# the fix is one identifier or one import rather than a judgement call.
_NAME_PATTERNS: tuple[tuple[str, re.Pattern[str]], ...] = (
    ("python", re.compile(r"NameError: name '(?P<symbol>[^']+)' is not defined")),
    ("python", re.compile(r"ModuleNotFoundError: No module named '(?P<symbol>[^']+)'")),
    (
        "python",
        re.compile(r"ImportError: cannot import name '(?P<symbol>[^']+)' from '(?P<origin>[^']+)'"),
    ),
    ("go", re.compile(r"^(?P<path>[\w./\-]+\.go):(?P<line>\d+):\d+:\s+undefined:\s+(?P<symbol>\w+)")),
    ("rust", re.compile(r"cannot find (?:value|function|type) `(?P<symbol>[^`]+)` in this scope")),
)
# The Python traceback names the file on the line above the error itself.
_PY_FRAME = re.compile(r'^\s*File "(?P<path>[^"]+)", line (?P<line>\d+)')
_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 extract_name_defects(bundle: dict, cfg: dict) -> list[dict[str, Any]]:
    """Return the undefined or misspelled names the build reported.

    Inputs: the evidence bundle and the per-job code cfg. Outputs: at most
    twenty {"category", "tool", "symbol", "path", "line", "message"} dicts.

    A Python traceback names the file on a frame line above the error, so the
    most recent frame is carried forward and attached to the error when it
    arrives. A defect whose file is unknown or outside the write allowlist is
    dropped, since the patcher could not act on it. Never raises.
    """

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


def extract_assertion_defects(bundle: dict, cfg: dict) -> list[dict[str, Any]]:
    """Return the failing tests the build published as structured results.

    Inputs: the evidence bundle, whose `jenkins.failed_tests` is populated when
    the build published test results, and the per-job code cfg. Outputs: at
    most twenty {"category", "test", "class", "message"} dicts.

    Carries no path on purpose: the failing test names the symptom, and the
    defect is usually in the code under test rather than in the test itself.
    Naming a path here would point the patcher at the test and invite it to
    weaken the assertion instead of fixing the cause. Never raises.
    """

    try:
        jenkins = bundle.get("jenkins") if isinstance(bundle.get("jenkins"), dict) else {}
        tests = jenkins.get("failed_tests")
        found = []
        for test in (tests if isinstance(tests, list) else [])[:_MAX_DEFECTS]:
            if not isinstance(test, dict) or not str(test.get("name") or "").strip():
                continue
            found.append(
                {
                    "category": FAILING_ASSERTION,
                    "test": str(test.get("name")).strip(),
                    "class": str(test.get("className") or "").strip(),
                    "message": " ".join(str(test.get("errorDetails") or "").split())[:400],
                }
            )
        return found
    except Exception:
        return []


def extract_sonar_defects(bundle: dict, cfg: dict) -> list[dict[str, Any]]:
    """Return the SonarQube findings carried on the bundle.

    Inputs: the evidence bundle, whose `sonarqube.issues` is populated by the
    quality sweep, and the per-job code cfg. Outputs: at most twenty
    {"category", "tool", "path", "line", "rule", "message"} dicts.

    Already normalized by the client and already located to a file and line,
    so this only enforces the write allowlist - the same boundary every other
    category is held to, applied again here because the findings arrive from
    outside the build. Never raises.
    """

    try:
        sonar = bundle.get("sonarqube") if isinstance(bundle.get("sonarqube"), dict) else {}
        issues = sonar.get("issues")
        found = []
        for issue in (issues if isinstance(issues, list) else [])[:_MAX_DEFECTS]:
            if not isinstance(issue, dict):
                continue
            path = str(issue.get("path") or "")
            if not path or not _is_writable(path, cfg):
                continue
            found.append(
                {
                    "category": SONARQUBE_ISSUE,
                    "tool": "sonarqube",
                    "path": path,
                    "line": issue.get("line"),
                    "rule": str(issue.get("rule") or ""),
                    "message": str(issue.get("message") or ""),
                }
            )
        return found
    except Exception:
        return []


def _scan_names(lines: list[str], cfg: dict) -> list[dict[str, Any]]:
    """Match undefined-name shapes, carrying the last traceback frame."""

    found: list[dict[str, Any]] = []
    seen: set[tuple[str, str]] = set()
    frame_path, frame_line = "", 0
    for raw in lines:
        line = raw.rstrip()
        frame = _PY_FRAME.match(line)
        if frame:
            frame_path, frame_line = frame.group("path"), int(frame.group("line"))
            continue
        for tool, pattern in _NAME_PATTERNS:
            match = pattern.search(line)
            if not match:
                continue
            groups = match.groupdict()
            path = groups.get("path") or frame_path
            line_no = int(groups["line"]) if groups.get("line") else frame_line
            if not path or not _is_writable(path, cfg):
                break
            key = (path, groups["symbol"])
            if key in seen:
                break
            seen.add(key)
            found.append(
                {
                    "category": UNDEFINED_NAME,
                    "tool": tool,
                    "symbol": groups["symbol"],
                    "path": path,
                    "line": line_no,
                    "message": line.strip()[:200],
                }
            )
            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)


ALL_CATEGORIES: tuple[str, ...] = (
    SONARQUBE_ISSUE,
    LINT_VIOLATION,
    UNDEFINED_NAME,
    FAILING_ASSERTION,
)


def enabled_categories(cfg: dict) -> tuple[str, ...]:
    """Return the defect categories this deployment allows Hermes to be asked about.

    Inputs: the per-job code cfg, whose `fix_categories` names the enabled
    ones. Outputs: the recognised subset, in declaration order.

    This is the operator's control over what Hermes may be pointed at. It is
    deliberately separate from the action allowlist: that gates what Ariadne
    executes on its own, whereas everything here becomes a pull request a
    person reads. An unrecognised name is ignored rather than trusted, and an
    empty setting means every category, so the default stays obvious.
    """

    raw = cfg.get("fix_categories") if isinstance(cfg, dict) else None
    names = {str(item).strip() for item in (raw or []) if str(item).strip()}
    if not names:
        return ALL_CATEGORIES
    return tuple(category for category in ALL_CATEGORIES if category in names)


def extract_defects(bundle: dict, cfg: dict) -> list[dict[str, Any]]:
    """Return every enabled mechanically-fixable defect the evidence names.

    Inputs: the evidence bundle and the per-job code cfg. Outputs: the enabled
    categories' defects together, most precisely located first, so the
    instruction leads with the defect whose fix is least open to
    interpretation.
    """

    enabled = enabled_categories(cfg)
    found: list[dict[str, Any]] = []
    if SONARQUBE_ISSUE in enabled:
        found += extract_sonar_defects(bundle, cfg)
    if LINT_VIOLATION in enabled:
        found += extract_lint_defects(bundle, cfg)
    if UNDEFINED_NAME in enabled:
        found += extract_name_defects(bundle, cfg)
    if FAILING_ASSERTION in enabled:
        found += extract_assertion_defects(bundle, cfg)
    return found


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_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, symbol or assertion is the point: it
    converts "repair this build" into a change whose correctness can be read
    off the same evidence.
    """

    if not defects:
        return ""
    # A static-analysis sweep runs against a green build. Saying "the build
    # reported" there would be a plain falsehood in the first line the model
    # reads, and the instruction two lines later already contradicts it.
    from_build = any(defect["category"] != SONARQUBE_ISSUE for defect in defects)
    lead = (
        "The build reported these specific defects; address exactly these:"
        if from_build
        else "Static analysis reported these findings; address exactly these:"
    )
    lines: list[str] = [lead]
    for defect in defects:
        lines.append(_defect_line(defect))
    if any(d["category"] == LINT_VIOLATION for d in defects):
        lines.append("Do not reformat unrelated lines and do not suppress the rule.")
    if any(d["category"] == SONARQUBE_ISSUE for d in defects):
        lines.append(
            "These are static-analysis findings, so the build is not failing. Preserve the "
            "existing behaviour exactly: fix what the rule objects to without changing what "
            "the code does, and do not suppress the rule or add an inline ignore."
        )
    if any(d["category"] == FAILING_ASSERTION for d in defects):
        lines.append(
            "Fix the code under test so the assertion holds. Do not weaken, skip or delete "
            "the test, and do not change its expected values."
        )
    return "\n".join(lines)


def _defect_line(defect: dict[str, Any]) -> str:
    """Render one defect as a single instruction bullet."""

    category = defect["category"]
    if category == SONARQUBE_ISSUE:
        return (
            f"- {defect['path']} line {defect['line']}: {defect['rule']} {defect['message']}"
        )
    if category == LINT_VIOLATION:
        return f"- {defect['path']} line {defect['line']}: {defect['rule']} {defect['message']}"
    if category == UNDEFINED_NAME:
        return (
            f"- {defect['path']} line {defect['line']}: the name {defect['symbol']!r} is not "
            f"defined or not importable ({defect['message']})"
        )
    return f"- test {defect['class']}.{defect['test']} failed: {defect['message']}"