diff --git a/ariadne/services/hermes_code_defects.py b/ariadne/services/hermes_code_defects.py new file mode 100644 index 0000000..f8129ef --- /dev/null +++ b/ariadne/services/hermes_code_defects.py @@ -0,0 +1,155 @@ +"""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[\w./\-]+\.py):(?P\d+):\d+:\s+(?P[A-Z]+\d+)\s+(?P.+)$" + ), + ), + # eslint (stylish): 12:5 error Unexpected console statement no-console + ( + "eslint", + re.compile( + r"^\s*(?P\d+):\d+\s+(?:error|warning)\s+(?P.+?)\s\s+(?P[\w\-/]+)$" + ), + ), + # golangci-lint: path.go:12:5: msg (govet) + ( + "golangci", + re.compile( + r"^(?P[\w./\-]+\.go):(?P\d+):\d+:\s+(?P.+?)\s+\((?P[\w\-]+)\)$" + ), + ), +) + +_ESLINT_FILE = re.compile(r"^(?P[\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) diff --git a/ariadne/services/hermes_code_flow.py b/ariadne/services/hermes_code_flow.py index b2842a4..b45eecb 100644 --- a/ariadne/services/hermes_code_flow.py +++ b/ariadne/services/hermes_code_flow.py @@ -25,6 +25,7 @@ from ..utils.logging import get_logger from . import ( hermes_agent_client, hermes_code_candidates, + hermes_code_prompt, hermes_code_patch, hermes_code_repair, hermes_code_repos, @@ -44,25 +45,6 @@ DEFAULT_MAX_OPEN_PROPOSALS = 64 _NO_REPO_MAPPING_REASON = "no_repo_mapping" _NO_CANDIDATE_FILES_REASON = "no_candidate_files" -_PATCH_PROMPT_TEMPLATE = """Use $triage-titan-test-failures. -You are proposing a MINIMAL source fix for incident __INCIDENT_ID__. -The repository is __OWNER__/__REPO__ branch __BASE_BRANCH__. -Return ONLY a single JSON object with exactly these keys and no others: -{"incident_id": "", "analysis": "", "patch": {"path": "", "original": "", "replacement": "", "rationale": ""} or null, "human_required": , "reason": ""} -`patch.path` MUST be exactly one of these candidate paths, copied character for character: -__CANDIDATE_LIST__ -`original` must be an exact substring of the content shown below for THAT file, appearing exactly once. -Change as few lines as possible; do not reformat; do not add dependencies. -Ariadne validates and pushes the change — you do not execute anything. -Set human_required to true if the fix is not a small localized source change. - -Failing test evidence bundle: -__BUNDLE__ - -__FILE_SECTIONS__""" - -_FILE_SECTION_TEMPLATE = "Current content of the candidate file {path}:\n{contents}" - @dataclass(frozen=True) class _Incident: @@ -273,7 +255,7 @@ def _diagnose( # noqa: PLR0913 - diagnosis needs both configs plus the fetched ) -> tuple[dict[str, Any], dict[str, Any]]: """Ask Hermes for a patch and gate it against the file it names.""" - prompt = _build_patch_prompt(incident.incident_id, bundle, cfg, fetched) + prompt = hermes_code_prompt.build_patch_prompt(incident.incident_id, bundle, cfg, fetched) run = hermes_agent_client.run_triage(hermes_cfg, prompt) if run.status != _RUN_COMPLETED or not run.output: return _human_required(f"hermes_run_{run.status}", run.run_id, validated=False) @@ -473,25 +455,3 @@ def _human_required( "url": None, } return result, event - - -def _build_patch_prompt( - incident_id: str, bundle: dict, code_cfg: dict, fetched: dict[str, str] -) -> str: - """Render the frozen patch prompt with incident, bundle, and file context.""" - - compact = json.dumps(bundle, separators=(",", ":"), ensure_ascii=True) - listing = "\n".join(f"- {path}" for path in fetched) - sections = "\n\n".join( - _FILE_SECTION_TEMPLATE.format(path=path, contents=contents) - for path, contents in fetched.items() - ) - return ( - _PATCH_PROMPT_TEMPLATE.replace("__INCIDENT_ID__", incident_id) - .replace("__OWNER__", str(code_cfg.get("owner") or "")) - .replace("__REPO__", str(code_cfg.get("repo") or "")) - .replace("__BASE_BRANCH__", str(code_cfg.get("base_branch") or "")) - .replace("__CANDIDATE_LIST__", listing) - .replace("__BUNDLE__", compact) - .replace("__FILE_SECTIONS__", sections) - ) diff --git a/ariadne/services/hermes_code_prompt.py b/ariadne/services/hermes_code_prompt.py new file mode 100644 index 0000000..db8c463 --- /dev/null +++ b/ariadne/services/hermes_code_prompt.py @@ -0,0 +1,58 @@ +"""Frozen patch prompt sent to Hermes when proposing a source fix. + +Separated from the flow so its wording is reviewed as a unit, and so a +linter's own diagnosis can be handed over verbatim: naming the file, line +and rule turns an open-ended repair request into a narrow instruction. +""" + +from __future__ import annotations + +import json + +from . import hermes_code_defects + + +_PATCH_PROMPT_TEMPLATE = """Use $triage-titan-test-failures. +You are proposing a MINIMAL source fix for incident __INCIDENT_ID__. +The repository is __OWNER__/__REPO__ branch __BASE_BRANCH__. +Return ONLY a single JSON object with exactly these keys and no others: +{"incident_id": "", "analysis": "", "patch": {"path": "", "original": "", "replacement": "", "rationale": ""} or null, "human_required": , "reason": ""} +`patch.path` MUST be exactly one of these candidate paths, copied character for character: +__CANDIDATE_LIST__ +`original` must be an exact substring of the content shown below for THAT file, appearing exactly once. +Change as few lines as possible; do not reformat; do not add dependencies. +Ariadne validates and pushes the change — you do not execute anything. +Set human_required to true if the fix is not a small localized source change. +__DEFECT_INSTRUCTION__ +Failing test evidence bundle: +__BUNDLE__ + +__FILE_SECTIONS__""" + +_FILE_SECTION_TEMPLATE = "Current content of the candidate file {path}:\n{contents}" + + +def build_patch_prompt( + incident_id: str, bundle: dict, code_cfg: dict, fetched: dict[str, str] +) -> str: + """Render the frozen patch prompt with incident, bundle, and file context.""" + + compact = json.dumps(bundle, separators=(",", ":"), ensure_ascii=True) + listing = "\n".join(f"- {path}" for path in fetched) + instruction = hermes_code_defects.defect_instruction( + hermes_code_defects.extract_lint_defects(bundle, code_cfg) + ) + sections = "\n\n".join( + _FILE_SECTION_TEMPLATE.format(path=path, contents=contents) + for path, contents in fetched.items() + ) + return ( + _PATCH_PROMPT_TEMPLATE.replace("__INCIDENT_ID__", incident_id) + .replace("__OWNER__", str(code_cfg.get("owner") or "")) + .replace("__REPO__", str(code_cfg.get("repo") or "")) + .replace("__BASE_BRANCH__", str(code_cfg.get("base_branch") or "")) + .replace("__CANDIDATE_LIST__", listing) + .replace("__DEFECT_INSTRUCTION__", f"{instruction}\n" if instruction else "") + .replace("__BUNDLE__", compact) + .replace("__FILE_SECTIONS__", sections) + ) diff --git a/tests/test_hermes_code_defects.py b/tests/test_hermes_code_defects.py new file mode 100644 index 0000000..68da59e --- /dev/null +++ b/tests/test_hermes_code_defects.py @@ -0,0 +1,122 @@ +"""Tests for recognising mechanically-fixable defects in console evidence.""" + +from __future__ import annotations + +from ariadne.services import hermes_code_defects as module + + +CFG = {"allowed_path_prefixes": ["ariadne/", "src/"], "allowed_suffixes": [".py", ".go", ".ts"]} + + +def _bundle(*lines: str) -> dict: + return {"jenkins": {"console_failures": [{"text": "\n".join(lines)}], "console_tail": ""}} + + +def test_ruff_diagnostic_is_extracted_whole() -> None: + found = module.extract_lint_defects( + _bundle("ariadne/app.py:12:5: E501 Line too long (95 > 88)"), CFG + ) + assert found == [ + { + "category": module.LINT_VIOLATION, + "tool": "ruff", + "path": "ariadne/app.py", + "line": 12, + "rule": "E501", + "message": "Line too long (95 > 88)", + } + ] + + +def test_golangci_diagnostic_is_extracted() -> None: + found = module.extract_lint_defects(_bundle("src/main.go:9:2: unused variable x (govet)"), CFG) + assert found[0]["tool"] == "golangci" + assert (found[0]["path"], found[0]["line"], found[0]["rule"]) == ("src/main.go", 9, "govet") + + +def test_eslint_diagnostics_inherit_the_preceding_file_header() -> None: + """eslint prints the path once, then indented diagnostics beneath it.""" + + found = module.extract_lint_defects( + _bundle("src/panel.ts", " 12:5 error Unexpected console statement no-console"), CFG + ) + assert (found[0]["path"], found[0]["line"], found[0]["rule"]) == ("src/panel.ts", 12, "no-console") + + +def test_a_file_outside_the_write_allowlist_is_dropped() -> None: + """Reporting a defect the patcher could never write is worse than silence.""" + + assert module.extract_lint_defects(_bundle("vendor/x.py:1:1: E501 Line too long"), CFG) == [] + assert module.extract_lint_defects(_bundle("ariadne/x.rb:1:1: E501 Line too long"), CFG) == [] + + +def test_duplicate_diagnostics_are_collapsed() -> None: + line = "ariadne/app.py:12:5: E501 Line too long (95 > 88)" + assert len(module.extract_lint_defects(_bundle(line, line, line), CFG)) == 1 + + +def test_extraction_is_bounded() -> None: + lines = [f"ariadne/f{i}.py:{i}:1: E501 Line too long" for i in range(50)] + assert len(module.extract_lint_defects(_bundle(*lines), CFG)) == module._MAX_DEFECTS + + +def test_ordinary_console_noise_matches_nothing() -> None: + noise = _bundle("Running tests...", "OK", "+ ruff check ariadne", "All checks passed!") + assert module.extract_lint_defects(noise, CFG) == [] + + +def test_never_raises_on_hostile_input() -> None: + for bad in (None, {}, {"jenkins": None}, {"jenkins": {"console_failures": "no"}}, 7): + assert module.extract_lint_defects(bad, CFG) == [] + assert module.extract_lint_defects(_bundle("ariadne/a.py:1:1: E1 x"), None) == [] + + +def test_instruction_names_rule_and_line() -> None: + defects = module.extract_lint_defects(_bundle("ariadne/app.py:12:5: E501 Line too long"), CFG) + text = module.defect_instruction(defects) + assert "ariadne/app.py line 12: E501" in text + assert "do not suppress the rule" in text + + +def test_no_defects_yields_no_instruction() -> None: + """The caller must fall back to the open-ended request, not an empty header.""" + + assert module.defect_instruction([]) == "" + + +def test_the_patch_prompt_carries_the_linter_diagnosis(monkeypatch) -> None: + """The instruction must reach Hermes, not just be computed.""" + + from ariadne.services import hermes_code_prompt + + bundle = { + "jenkins": { + "console_failures": [{"text": "ariadne/app.py:12:5: E501 Line too long (95 > 88)"}], + "console_tail": "", + } + } + cfg = { + "owner": "bstein", + "repo": "ariadne", + "base_branch": "master", + "allowed_path_prefixes": ["ariadne/"], + "allowed_suffixes": [".py"], + } + prompt = hermes_code_prompt.build_patch_prompt("ariadne/1", bundle, cfg, {"ariadne/app.py": "x"}) + + assert "ariadne/app.py line 12: E501" in prompt + assert "do not suppress the rule" in prompt + + +def test_the_patch_prompt_is_unchanged_without_diagnostics() -> None: + """A build with no linter output must fall back to the open request.""" + + from ariadne.services import hermes_code_prompt + + bundle = {"jenkins": {"console_failures": [{"text": "boom"}], "console_tail": ""}} + cfg = {"owner": "b", "repo": "r", "base_branch": "main", + "allowed_path_prefixes": ["ariadne/"], "allowed_suffixes": [".py"]} + prompt = hermes_code_prompt.build_patch_prompt("a/1", bundle, cfg, {"ariadne/a.py": "x"}) + + assert "A linter reported these violations" not in prompt + assert "MINIMAL source fix" in prompt