diff --git a/ariadne/services/hermes_sonar_advice.py b/ariadne/services/hermes_sonar_advice.py new file mode 100644 index 0000000..ae44c07 --- /dev/null +++ b/ariadne/services/hermes_sonar_advice.py @@ -0,0 +1,249 @@ +"""File an issue with a suggested fix when a finding cannot become a patch. + +The sweep opens a pull request when everything lines up: the project maps to a +repository, the file sits inside the write allowlist, and the change is one +anchored snippet the validator can check. Plenty of real findings fail one of +those and produce nothing at all - which is the wrong outcome, because a +finding nobody can patch automatically is exactly the one a maintainer has to +do by hand, and that is when knowing the intended fix is worth most. + +So the fallback is an issue carrying the finding and the code Hermes believes +would resolve it. Not a patch: nothing here is anchored, validated against the +file, or pushed. It is written for a person who will read it, judge it, and +make the change themselves, which is why it may describe work too large or too +diffuse for the patcher to have attempted. + +Deduped on the rule, like the pull requests. One rule is one root cause spread +over many files, and an issue per instance would bury the repository in near +identical tickets - the precise failure this whole path exists to avoid. +""" + +from __future__ import annotations + +import json +from typing import Any + +from ..utils.logging import get_logger +from . import ( + hermes_agent_client, + hermes_code_repair, + hermes_code_suggestion, + hermes_incident_issue, + hermes_sonar_client, +) + + +logger = get_logger(__name__) + +ADVICE_EVENT_TYPE = "hermes_sonar_advice" + +_RUN_COMPLETED = "completed" +_MAX_CONTEXT_CHARS = 40000 + +_PROMPT = """A static-analysis finding, not a build failure. Nothing is broken and no investigation is needed: the finding and the file are below. +You are suggesting a fix for a maintainer to make by hand. Ariadne cannot apply this automatically, so it will be printed in an issue for a person to read. +Return ONLY a single JSON object with exactly these keys and no others: +{"incident_id": "", "analysis": "", "code_suggestions": [{"path": "", "explanation": "", "code": ""}], "human_required": , "reason": ""} +Give at most three suggestions and prefer one. 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. +Nothing you return is applied, validated against the file, or committed anywhere, so suggest the fix you actually believe is right even when it spans several places. + +The finding: +__FINDING__ + +Current content of __PATH__: +__CONTENTS__""" + + +def advise( # noqa: PLR0913 - one advice attempt needs the whole proposal context + storage: Any, + config: Any, + hermes_cfg: dict, + repo_cfg: dict, + project: str, + issue: dict[str, Any], +) -> dict[str, Any]: + """Ask Hermes for a fix and file it as an issue for one finding. + + Inputs: event storage; a settings-like object; the Hermes run config; the + resolved per-repo cfg; the SonarQube project key; and one normalized + finding. Outputs: {"filed": bool, "reason": str, "url": str|None}. + + Never raises. This runs after a pull request has already been declined, so + a failure here must not turn a partial outcome into no outcome at all. + """ + + try: + return _advise(storage, config, hermes_cfg, repo_cfg, project, issue) + except Exception as exc: + logger.info( + "hermes sonar advice failed", + extra={"event": ADVICE_EVENT_TYPE, "status": "error", "detail": str(exc)}, + ) + return {"filed": False, "reason": f"advice_failed: {exc}", "url": None} + + +def _advise( # noqa: PLR0913 - mirrors advise's contract + storage: Any, + config: Any, + hermes_cfg: dict, + repo_cfg: dict, + project: str, + issue: dict[str, Any], +) -> dict[str, Any]: + """Run the advice flow, recording one event whatever the outcome.""" + + incident_id = f"{hermes_sonar_client.INCIDENT_PREFIX}{project}/{issue.get('rule')}/{issue.get('key')}" + rule = str(issue.get("rule") or "unknown_rule") + issue_cfg = _issue_config(config, repo_cfg) + + existing = hermes_incident_issue.find_open_incident_issue(issue_cfg, project, rule, incident_id) + if existing.get("found"): + return _record(storage, incident_id, {"filed": False, "reason": "issue_already_open", "url": existing.get("url")}) + + contents, error = hermes_code_repair.fetch_file(repo_cfg, str(issue.get("path") or "")) + if contents is None: + return _record(storage, incident_id, {"filed": False, "reason": f"file_fetch_failed: {error}", "url": None}) + + run = hermes_agent_client.run_triage(hermes_cfg, _prompt(incident_id, issue, contents)) + if run.status != _RUN_COMPLETED or not run.output: + return _record(storage, incident_id, {"filed": False, "reason": f"hermes_run_{run.status}", "url": None}) + + suggestions, analysis = parse_advice(run.output, incident_id) + if not suggestions: + return _record(storage, incident_id, {"filed": False, "reason": "no_suggestions_returned", "url": None}) + + created = hermes_incident_issue.create_incident_issue( + issue_cfg, _context(project, rule, incident_id, issue, analysis, suggestions, run.run_id, config) + ) + if created.get("error"): + return _record(storage, incident_id, {"filed": False, "reason": str(created["error"]), "url": None}) + return _record(storage, incident_id, {"filed": True, "reason": "issue_filed", "url": created.get("url")}) + + +def parse_advice(raw_output: str, incident_id: str) -> tuple[list[Any], str]: + """Extract the suggestions and analysis from an advice response. + + Inputs: raw model output and the incident it must reference. Outputs: + (suggestions, analysis); suggestions is empty when the response is + unusable, which the caller treats as nothing to file. Never raises. + + Validated with the same rules the triage schema applies, so an issue can + never carry a shape the rest of the system would have rejected. + """ + + try: + payload = json.loads(_first_object(str(raw_output or "")) or "") + except Exception: + return [], "" + if not isinstance(payload, dict) or payload.get("incident_id") != incident_id: + return [], "" + if hermes_code_suggestion.validate(payload) is not None: + return [], "" + analysis = payload.get("analysis") + return hermes_code_suggestion.from_payload(payload), analysis if isinstance(analysis, str) else "" + + +def _first_object(raw: str) -> str | None: + """Return the first balanced top-level JSON object in raw text.""" + + depth = 0 + start = -1 + in_string = False + escaped = False + for index, char in enumerate(raw): + if in_string: + escaped = char == "\\" and not escaped + if char == '"' and not escaped: + in_string = False + continue + if char == '"' and depth > 0: + in_string = True + elif char == "{": + if depth == 0: + start = index + depth += 1 + elif char == "}" and depth > 0: + depth -= 1 + if depth == 0: + return raw[start : index + 1] + return None + + +def _prompt(incident_id: str, issue: dict[str, Any], contents: str) -> str: + """Render the advice prompt for one finding and its file.""" + + finding = json.dumps( + {key: issue.get(key) for key in ("rule", "severity", "type", "effort", "path", "line", "message")}, + separators=(",", ":"), + ) + return ( + _PROMPT.replace("__INCIDENT_ID__", incident_id) + .replace("__FINDING__", finding) + .replace("__PATH__", str(issue.get("path") or "")) + .replace("__CONTENTS__", contents[:_MAX_CONTEXT_CHARS]) + ) + + +def _context( # noqa: PLR0913 - the issue body needs every one of these + project: str, + rule: str, + incident_id: str, + issue: dict[str, Any], + analysis: str, + suggestions: list[Any], + run_id: str | None, + config: Any, +) -> dict[str, Any]: + """Build the issue context for one advised finding.""" + + ui_url = str(getattr(config, "hermes_ui_url", "") or "") + finding_url = hermes_sonar_client.issue_url( + incident_id, str(getattr(config, "hermes_sonar_ui_url", "") or "") + ) + reason = analysis or f"SonarQube reports {rule} in {issue.get('path')}." + return { + "incident_id": incident_id, + "job": project, + "build_number": issue.get("line"), + "build_url": finding_url, + "classification": rule, + "confidence": None, + "first_failed_gate": "", + "reason": reason, + # Says plainly why nobody patched it, so the issue does not read as a + # failure of the automation. + "authorize_reason": "no automated patch was possible for this finding", + "facts": [ + { + "statement": f"{issue.get('message')} ({issue.get('severity')}, {issue.get('effort')} estimated)", + "source": "gitea", + "reference": f"{issue.get('path')}:{issue.get('line')}", + } + ], + "inferences": [], + "code_suggestions": suggestions, + "run_id": run_id, + "run_url": hermes_code_repair.run_url(ui_url, str(run_id or "")), + } + + +def _issue_config(config: Any, repo_cfg: dict) -> dict[str, Any]: + """Build the Gitea cfg for filing into this finding's repository.""" + + return { + "gitea_base_url": repo_cfg.get("gitea_base_url") or getattr(config, "hermes_gitea_base_url", ""), + "gitea_token": repo_cfg.get("gitea_token") or getattr(config, "hermes_gitea_token", ""), + "owner": repo_cfg.get("owner"), + "repo": repo_cfg.get("repo"), + "timeout_seconds": repo_cfg.get("timeout_seconds") or 15.0, + } + + +def _record(storage: Any, incident_id: str, result: dict[str, Any]) -> dict[str, Any]: + """Record one advice attempt, whatever it decided.""" + + try: + storage.record_event(ADVICE_EVENT_TYPE, {"incident_id": incident_id, **result}) + except Exception: + pass + return result diff --git a/ariadne/services/hermes_sonar_sweep.py b/ariadne/services/hermes_sonar_sweep.py index 23d431c..9d73e38 100644 --- a/ariadne/services/hermes_sonar_sweep.py +++ b/ariadne/services/hermes_sonar_sweep.py @@ -27,7 +27,7 @@ from typing import Any from ..settings import settings from ..utils.logging import get_logger -from . import hermes_code_flow, hermes_code_repair, hermes_sonar_client +from . import hermes_code_flow, hermes_code_repair, hermes_sonar_advice, hermes_sonar_client logger = get_logger(__name__) @@ -84,7 +84,7 @@ def sweep(storage: Any, config: Any, hermes_cfg: dict) -> dict[str, Any]: "hermes sonar sweep failed", extra={"event": SWEEP_EVENT_TYPE, "status": "error", "detail": str(exc)}, ) - return {"proposed": 0, "skipped": [f"sweep_failed: {exc}"], "projects": 0} + return {"proposed": 0, "advised": 0, "skipped": [f"sweep_failed: {exc}"], "projects": 0} def _sweep(storage: Any, config: Any, hermes_cfg: dict) -> dict[str, Any]: @@ -92,11 +92,12 @@ def _sweep(storage: Any, config: Any, hermes_cfg: dict) -> dict[str, Any]: projects = _projects(config) if not projects: - return {"proposed": 0, "skipped": [_NO_PROJECTS], "projects": 0} + return {"proposed": 0, "advised": 0, "skipped": [_NO_PROJECTS], "projects": 0} sonar_cfg = client_config(config) code_cfg = hermes_code_flow.code_config(config) budget = _max_per_sweep(config) proposed = 0 + advised = 0 skipped: list[str] = [] for project, job in sorted(projects.items()): if proposed >= budget: @@ -106,12 +107,13 @@ def _sweep(storage: Any, config: Any, hermes_cfg: dict) -> dict[str, Any]: storage, project, job, sonar_cfg, code_cfg, hermes_cfg, config ) proposed += outcome["proposed"] + advised += outcome.get("advised", 0) skipped.extend(outcome["skipped"]) logger.info( "hermes sonar sweep finished", extra={"event": SWEEP_EVENT_TYPE, "status": "ok", "detail": f"proposed={proposed}"}, ) - return {"proposed": proposed, "skipped": skipped, "projects": len(projects)} + return {"proposed": proposed, "advised": advised, "skipped": skipped, "projects": len(projects)} def _propose_for_project( # noqa: PLR0913 - one project needs every config the flow does @@ -133,20 +135,73 @@ def _propose_for_project( # noqa: PLR0913 - one project needs every config the return {"proposed": 0, "skipped": [f"{project}: {error}"]} open_rules = proposed_rules(repo_cfg, project) chosen = select_issue(issues, repo_cfg, config, open_rules) - if chosen is None: - return {"proposed": 0, "skipped": [f"{project}: no new mechanically fixable finding"]} - result = hermes_code_flow.propose_code_fix( - storage, - incident_id=incident_id(project, chosen), - job=job, - build_number=f"sonar-{chosen['key']}", - bundle=bundle_for(project, chosen), - hermes_cfg=hermes_cfg, - code_cfg=code_cfg, - ) - if result.get("status") == "pr_opened": - return {"proposed": 1, "skipped": []} - return {"proposed": 0, "skipped": [f"{project}: {result.get('reason')}"]} + if chosen is not None: + result = hermes_code_flow.propose_code_fix( + storage, + incident_id=incident_id(project, chosen), + job=job, + build_number=f"sonar-{chosen['key']}", + bundle=bundle_for(project, chosen), + hermes_cfg=hermes_cfg, + code_cfg=code_cfg, + ) + if result.get("status") == "pr_opened": + return {"proposed": 1, "skipped": []} + # A declined pull request is not a dead end. The finding is still real + # and a person still has to deal with it, so fall through to advice. + return _advise(storage, config, hermes_cfg, repo_cfg, project, chosen, str(result.get("reason"))) + advisable = select_advice_issue(issues, repo_cfg, config, open_rules) + if advisable is None: + return {"proposed": 0, "skipped": [f"{project}: no new finding to act on"]} + return _advise(storage, config, hermes_cfg, repo_cfg, project, advisable, "too large to patch") + + +def _advise( # noqa: PLR0913 - the advice call needs the whole proposal context + storage: Any, + config: Any, + hermes_cfg: dict, + repo_cfg: dict, + project: str, + issue: dict, + why: str, +) -> dict[str, Any]: + """File an issue with a suggested fix for a finding no patch covered.""" + + if not _advice_enabled(config): + return {"proposed": 0, "skipped": [f"{project}: {why} (advice disabled)"]} + outcome = hermes_sonar_advice.advise(storage, config, hermes_cfg, repo_cfg, project, issue) + if outcome.get("filed"): + return {"proposed": 1, "skipped": [], "advised": 1} + return {"proposed": 0, "skipped": [f"{project}: {why} -> {outcome.get('reason')}"]} + + +def select_advice_issue( + issues: list[dict], repo_cfg: dict, config: Any, open_rules: set[str] | None = None +) -> dict[str, Any] | None: + """Pick the finding most worth explaining to a person. + + Inputs and ordering as `select_issue`, minus the effort ceiling: this path + exists precisely for the findings too large to patch, which on this + instance is most of the backlog. Still bounded to writable paths, because + advising on a file nobody would edit is noise. + """ + + seen = open_rules or set() + eligible = [ + issue + for issue in issues + if str(issue.get("rule") or "") not in seen + and _is_writable(str(issue.get("path") or ""), repo_cfg) + ] + if not eligible: + return None + return sorted(eligible, key=_selection_key)[0] + + +def _advice_enabled(config: Any) -> bool: + """Report whether findings with no patch may be filed as issues.""" + + return bool(getattr(config, "hermes_sonar_advice_enabled", False)) def proposed_rules(repo_cfg: dict, project: str) -> set[str]: diff --git a/ariadne/settings.py b/ariadne/settings.py index 05b340a..d535256 100644 --- a/ariadne/settings.py +++ b/ariadne/settings.py @@ -325,6 +325,7 @@ class Settings: hermes_sonar_types: list[str] hermes_sonar_severities: list[str] hermes_sonar_max_per_sweep: int + hermes_sonar_advice_enabled: bool hermes_sonar_max_effort_minutes: int hermes_sonar_timeout_seconds: float diff --git a/ariadne/settings_hermes.py b/ariadne/settings_hermes.py index fe5c772..103b2e5 100644 --- a/ariadne/settings_hermes.py +++ b/ariadne/settings_hermes.py @@ -73,6 +73,7 @@ def _hermes_autotriage_config() -> dict[str, Any]: if item.strip() ], "hermes_sonar_max_per_sweep": _env_int("ARIADNE_HERMES_SONAR_MAX_PER_SWEEP", 1), + "hermes_sonar_advice_enabled": _env_bool("ARIADNE_HERMES_SONAR_ADVICE_ENABLED", "false"), "hermes_sonar_max_effort_minutes": _env_int("ARIADNE_HERMES_SONAR_MAX_EFFORT_MINUTES", 20), "hermes_sonar_timeout_seconds": _env_float("ARIADNE_HERMES_SONAR_TIMEOUT_SECONDS", 20.0), "hermes_demo_namespace": _env("ARIADNE_HERMES_DEMO_NAMESPACE", "hermes-triage-demo"), diff --git a/tests/test_hermes_sonar_advice.py b/tests/test_hermes_sonar_advice.py new file mode 100644 index 0000000..93faece --- /dev/null +++ b/tests/test_hermes_sonar_advice.py @@ -0,0 +1,234 @@ +"""Tests for filing a suggested fix when a finding cannot become a patch.""" + +from __future__ import annotations + +import json +from types import SimpleNamespace + +import pytest + +from ariadne.services import hermes_sonar_advice as module + + +ISSUE = { + "key": "AZ1", + "rule": "python:S3776", + "severity": "CRITICAL", + "type": "CODE_SMELL", + "effort": "11min", + "path": "ariadne/services/thing.py", + "line": 42, + "message": "Refactor this function to reduce its Cognitive Complexity from 21 to the 15 allowed.", +} + +REPO_CFG = { + "owner": "bstein", + "repo": "ariadne", + "gitea_base_url": "https://scm.example", + "gitea_token": "t", + "base_branch": "master", +} + +INCIDENT = "sonar/ariadne/python:S3776/AZ1" + +SUGGESTION = { + "path": "ariadne/services/thing.py", + "explanation": "Extract the validation branch into its own helper.", + "code": "def _validate(row):\n return bool(row)", +} + + +def _config(**overrides): + values = { + "hermes_ui_url": "https://agent.example", + "hermes_sonar_ui_url": "https://quality.example", + "hermes_gitea_base_url": "https://scm.example", + "hermes_gitea_token": "t", + } + values.update(overrides) + return SimpleNamespace(**values) + + +class _Storage: + def __init__(self): + self.events = [] + + def record_event(self, event_type, detail): + self.events.append((event_type, detail)) + + +def _response(**overrides): + payload = { + "incident_id": INCIDENT, + "analysis": "The function branches five ways over the same row.", + "code_suggestions": [dict(SUGGESTION)], + "human_required": True, + "reason": "needs a maintainer", + } + payload.update(overrides) + return json.dumps(payload) + + +@pytest.fixture +def wiring(monkeypatch): + state = { + "existing": {"found": False}, + "contents": "def thing():\n pass\n", + "fetch_error": None, + "run": SimpleNamespace(status="completed", output=_response(), run_id="run_x"), + "created": {"issue_number": 9, "url": "https://scm.example/issues/9", "error": None}, + "filed_context": [], + } + monkeypatch.setattr( + module.hermes_incident_issue, "find_open_incident_issue", + lambda cfg, job, classification, incident: state["existing"], + ) + monkeypatch.setattr( + module.hermes_code_repair, "fetch_file", + lambda cfg, path: (state["contents"], state["fetch_error"]), + ) + monkeypatch.setattr(module.hermes_agent_client, "run_triage", lambda cfg, prompt: state["run"]) + + def _create(cfg, context): + state["filed_context"].append(context) + return state["created"] + + monkeypatch.setattr(module.hermes_incident_issue, "create_incident_issue", _create) + return state + + +def test_a_finding_becomes_an_issue_carrying_the_suggested_code(wiring) -> None: + storage = _Storage() + + result = module.advise(storage, _config(), {}, REPO_CFG, "ariadne", dict(ISSUE)) + + assert result == {"filed": True, "reason": "issue_filed", "url": "https://scm.example/issues/9"} + context = wiring["filed_context"][0] + assert context["incident_id"] == INCIDENT + assert context["classification"] == "python:S3776" + assert context["code_suggestions"][0].code.startswith("def _validate") + assert context["reason"].startswith("The function branches") + + +def test_the_issue_links_to_both_the_finding_and_the_run(wiring) -> None: + module.advise(_Storage(), _config(), {}, REPO_CFG, "ariadne", dict(ISSUE)) + + context = wiring["filed_context"][0] + assert context["build_url"] == ( + "https://quality.example/project/issues?resolved=false&id=ariadne&open=AZ1" + ) + assert context["run_url"] == "https://agent.example/chat?resume=run_x" + + +def test_the_issue_says_why_nothing_was_patched(wiring) -> None: + """Otherwise it reads as the automation having failed.""" + + module.advise(_Storage(), _config(), {}, REPO_CFG, "ariadne", dict(ISSUE)) + + assert "no automated patch was possible" in wiring["filed_context"][0]["authorize_reason"] + + +def test_a_rule_already_filed_is_not_filed_again(wiring) -> None: + """One rule is one root cause; an issue per instance buries the repo.""" + + wiring["existing"] = {"found": True, "url": "https://scm.example/issues/3"} + + result = module.advise(_Storage(), _config(), {}, REPO_CFG, "ariadne", dict(ISSUE)) + + assert result["filed"] is False + assert result["reason"] == "issue_already_open" + assert wiring["filed_context"] == [] + + +def test_an_unreadable_file_files_nothing(wiring) -> None: + wiring["contents"] = None + wiring["fetch_error"] = "http 404" + + result = module.advise(_Storage(), _config(), {}, REPO_CFG, "ariadne", dict(ISSUE)) + + assert result["filed"] is False + assert "file_fetch_failed" in result["reason"] + + +@pytest.mark.parametrize( + ("status", "output"), + [("failed", ""), ("completed", ""), ("timeout", "{}")], +) +def test_a_run_that_produced_nothing_files_nothing(wiring, status, output) -> None: + wiring["run"] = SimpleNamespace(status=status, output=output, run_id="r") + + assert module.advise(_Storage(), _config(), {}, REPO_CFG, "ariadne", dict(ISSUE))["filed"] is False + + +def test_a_response_with_no_suggestions_files_nothing(wiring) -> None: + wiring["run"] = SimpleNamespace( + status="completed", output=_response(code_suggestions=[]), run_id="r" + ) + + result = module.advise(_Storage(), _config(), {}, REPO_CFG, "ariadne", dict(ISSUE)) + + assert result["reason"] == "no_suggestions_returned" + + +def test_a_failed_creation_is_reported(wiring) -> None: + wiring["created"] = {"issue_number": None, "url": None, "error": "http 500"} + + assert module.advise(_Storage(), _config(), {}, REPO_CFG, "ariadne", dict(ISSUE))["filed"] is False + + +def test_every_attempt_is_recorded(wiring) -> None: + storage = _Storage() + module.advise(storage, _config(), {}, REPO_CFG, "ariadne", dict(ISSUE)) + + assert storage.events[0][0] == module.ADVICE_EVENT_TYPE + assert storage.events[0][1]["incident_id"] == INCIDENT + + +def test_advice_never_raises(monkeypatch) -> None: + """It runs after a pull request was already declined; it must not erase it.""" + + monkeypatch.setattr( + module.hermes_incident_issue, "find_open_incident_issue", + lambda *a, **k: (_ for _ in ()).throw(RuntimeError("boom")), + ) + + result = module.advise(_Storage(), _config(), {}, REPO_CFG, "ariadne", dict(ISSUE)) + + assert result["filed"] is False + assert "advice_failed" in result["reason"] + + +@pytest.mark.parametrize( + "raw", + [ + "", + "not json at all", + '{"incident_id": "someone/else", "analysis": "a", "code_suggestions": [], "human_required": true, "reason": "r"}', + '{"incident_id": "' + INCIDENT + '", "code_suggestions": "nope"}', + ], +) +def test_an_unusable_response_yields_no_suggestions(raw) -> None: + assert module.parse_advice(raw, INCIDENT) == ([], "") + + +def test_a_good_response_is_parsed() -> None: + suggestions, analysis = module.parse_advice("noise " + _response() + " trailing", INCIDENT) + + assert len(suggestions) == 1 + assert suggestions[0].path == "ariadne/services/thing.py" + assert analysis.startswith("The function branches") + + +def test_the_prompt_states_that_nothing_is_applied() -> None: + prompt = module._prompt(INCIDENT, dict(ISSUE), "def thing(): pass") + + assert "not a build failure" in prompt + assert "Nothing you return is applied" in prompt + assert "python:S3776" in prompt + assert "def thing(): pass" in prompt + + +def test_the_prompt_bounds_the_file_it_sends() -> None: + prompt = module._prompt(INCIDENT, dict(ISSUE), "x" * 90000) + + assert len(prompt) < 45000 diff --git a/tests/test_hermes_sonar_sweep.py b/tests/test_hermes_sonar_sweep.py index 45a737a..8dcd8f6 100644 --- a/tests/test_hermes_sonar_sweep.py +++ b/tests/test_hermes_sonar_sweep.py @@ -42,6 +42,7 @@ def _config(**overrides): "hermes_sonar_max_per_sweep": 1, "hermes_sonar_max_effort_minutes": 20, "hermes_sonar_timeout_seconds": 5, + "hermes_sonar_advice_enabled": False, } values.update(overrides) return SimpleNamespace(**values) @@ -148,7 +149,7 @@ def test_a_sweep_opens_one_proposal_for_the_chosen_finding(wiring) -> None: result = module.sweep(storage, _config(), {}) - assert result == {"proposed": 1, "skipped": [], "projects": 1} + assert result == {"proposed": 1, "advised": 0, "skipped": [], "projects": 1} proposal = wiring["proposals"][0] assert proposal["incident_id"] == "sonar/ariadne/python:S1172/AZ-1" assert proposal["job"] == "ariadne" @@ -186,13 +187,13 @@ def test_a_declined_proposal_is_reported_and_does_not_spend_budget(wiring) -> No assert result["proposed"] == 0 assert len(wiring["proposals"]) == 2 - assert "a: open_proposal_limit_reached" in result["skipped"] + assert any("open_proposal_limit_reached" in item for item in result["skipped"]) def test_no_configured_projects_makes_no_call(wiring) -> None: result = module.sweep(_Storage(), _config(hermes_sonar_projects={}), {}) - assert result == {"proposed": 0, "skipped": ["no_projects_configured"], "projects": 0} + assert result == {"proposed": 0, "advised": 0, "skipped": ["no_projects_configured"], "projects": 0} assert wiring["proposals"] == [] @@ -215,12 +216,16 @@ def test_a_fetch_error_is_reported_not_raised(wiring) -> None: assert result["skipped"] == ["ariadne: sonar fetch http 503"] -def test_a_project_with_nothing_mechanical_is_reported(wiring) -> None: +def test_a_finding_too_large_to_patch_becomes_advice(wiring) -> None: + """The bulk of the backlog is refactors no anchored patch can express.""" + wiring["issues"] = [_issue(effort="8h")] result = module.sweep(_Storage(), _config(), {}) - assert result["skipped"] == ["ariadne: no new mechanically fixable finding"] + assert result["proposed"] == 0 + assert result["skipped"] == ["ariadne: too large to patch (advice disabled)"] + assert wiring["proposals"] == [] def test_a_sweep_never_takes_down_the_scheduler(monkeypatch) -> None: @@ -388,7 +393,7 @@ def test_every_rule_under_review_means_nothing_new_to_propose(wiring) -> None: result = module.sweep(_Storage(), _config(), {}) assert result["proposed"] == 0 - assert result["skipped"] == ["ariadne: no new mechanically fixable finding"] + assert result["skipped"] == ["ariadne: no new finding to act on"] def test_another_projects_open_proposal_does_not_block_this_one(wiring) -> None: @@ -423,3 +428,50 @@ def test_an_unreadable_proposal_list_costs_a_duplicate_not_a_silent_rule(monkeyp ) assert module.proposed_rules(REPO_CFG, "ariadne") == set() + + +def test_advice_files_an_issue_when_no_patch_was_possible(monkeypatch, wiring) -> None: + """A finding nobody can patch is exactly the one worth explaining.""" + + filed = [] + monkeypatch.setattr( + module.hermes_sonar_advice, "advise", + lambda storage, config, hermes_cfg, repo_cfg, project, issue: filed.append(issue) + or {"filed": True, "reason": "issue_filed", "url": "https://scm/issues/9"}, + ) + wiring["issues"] = [_issue(effort="8h", key="big")] + + result = module.sweep(_Storage(), _config(hermes_sonar_advice_enabled=True), {}) + + assert result["advised"] == 1 + assert filed[0]["key"] == "big" + + +def test_a_declined_pull_request_falls_through_to_advice(monkeypatch, wiring) -> None: + """A declined proposal is not a dead end; the finding is still real.""" + + monkeypatch.setattr( + module.hermes_sonar_advice, "advise", + lambda *a, **k: {"filed": True, "reason": "issue_filed", "url": "u"}, + ) + wiring["result"] = {"status": "human_required", "reason": "patch_rejected: original_ambiguous"} + + result = module.sweep(_Storage(), _config(hermes_sonar_advice_enabled=True), {}) + + assert result["advised"] == 1 + + +def test_advice_respects_the_rules_already_under_review(wiring) -> None: + wiring["open_incidents"] = ["sonar/ariadne/python:S1172/OTHER"] + wiring["issues"] = [_issue(effort="8h", rule="python:S1172")] + + result = module.sweep(_Storage(), _config(hermes_sonar_advice_enabled=True), {}) + + assert result["skipped"] == ["ariadne: no new finding to act on"] + + +def test_advice_stays_inside_the_write_allowlist() -> None: + """Advising on a file nobody would edit is noise, not help.""" + + assert module.select_advice_issue([_issue(path="docs/x.md", effort="8h")], REPO_CFG, _config()) is None + assert module.select_advice_issue([_issue(effort="8h")], REPO_CFG, _config())["key"] == "AZ-1"