feat(hermes): propose fixes for SonarQube findings on a schedule

Triage has only ever entered on a failure. Static analysis is the opposite
shape - a standing backlog that never fails a build and so never asks anyone
for attention. On this instance that backlog is 139 open findings on Ariadne
alone, each already naming its file, its line, its rule and what is wrong.
That is better-located evidence than the console text the code-repair flow
normally mines, and it was being thrown away.

This is a second way into the same flow, not a second flow. A scheduled sweep
picks one finding and hands it to the existing proposal path, which is
unchanged: Hermes returns a patch as data, Ariadne validates it against the
file it names, pushes a branch, opens a pull request nobody merges. A finding
arriving from outside the build is not a reason to relax the gates that make a
proposal worth reading, so it does not.

Three deliberate limits. Security hotspots are never fetched: SonarQube models
them as needing human review, the quality gate here fails on exactly that
condition, and an automation that resolved them would be marking them
reviewed without review - defeating the control rather than satisfying it.
Findings already marked won't-fix carry a judgement someone made, and
reopening it produces pull requests that argue with a person. And the sweep
proposes one fix per run by default, because 139 pull requests nobody reads
would make the review gate theatre.

Selection is by SonarQube's own effort estimate rather than severity: effort
is the closest available proxy for the one-anchor change the patch validator
can actually check, so a trivial CRITICAL beats an involved MINOR. An
unparseable estimate is treated as ineligible, not as free.

Off by default. Triage reacts to a failure someone already cares about; this
opens pull requests nobody asked for, and that is a decision an operator makes
deliberately rather than inherits on upgrade.

Branch naming now sanitizes its token, since it arrives from a finding key as
well as a build number and a ref is one of the few places where an unexpected
character stops being cosmetic.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

ariadne/app.py

@@ -23,6 +23,7 @@ from .services.firefly import firefly
 from .services.game_mode import game_mode
 from .services.game_stream_profiles import game_stream_profiles
 from .services.hermes_autotriage import run_hermes_autotriage
+from .services.hermes_sonar_sweep import run_hermes_sonar_sweep
 from .services.image_sweeper import image_sweeper
 from .services.jenkins_build_weather import collect_jenkins_build_weather
 from .services.jenkins_workspace_cleanup import cleanup_jenkins_workspace_storage
@@ -192,6 +193,7 @@ def _startup() -> None:
     scheduler.add_task("schedule.jenkins_workspace_cleanup", settings.jenkins_workspace_cleanup_cron, cleanup_jenkins_workspace_storage)
     scheduler.add_task("schedule.testing_triage", settings.testing_triage_cron, lambda: run_testing_triage(storage))
     scheduler.add_task("schedule.hermes_autotriage", settings.hermes_autotriage_cron, lambda: run_hermes_autotriage(storage))
+    scheduler.add_task("schedule.hermes_sonar_sweep", settings.hermes_sonar_cron, lambda: run_hermes_sonar_sweep(storage))
     scheduler.add_task("schedule.vault_k8s_auth", settings.vault_k8s_auth_cron, lambda: vault.sync_k8s_auth(wait=True))
     scheduler.add_task("schedule.vault_oidc", settings.vault_oidc_cron, lambda: vault.sync_oidc(wait=True))
     scheduler.add_task("schedule.comms_guest_name", settings.comms_guest_name_cron, lambda: comms.run_guest_name_randomizer(wait=True))
@@ -227,6 +229,7 @@ def _startup() -> None:
             "jenkins_workspace_cleanup_max_deletions_per_run": settings.jenkins_workspace_cleanup_max_deletions_per_run,
             "testing_triage_cron": settings.testing_triage_cron,
             "hermes_autotriage_cron": settings.hermes_autotriage_cron,
+            "hermes_sonar_cron": settings.hermes_sonar_cron,
             "vault_k8s_auth_cron": settings.vault_k8s_auth_cron,
             "vault_oidc_cron": settings.vault_oidc_cron,
             "comms_guest_name_cron": settings.comms_guest_name_cron,

ariadne/services/hermes_code_candidates.py

@@ -83,12 +83,42 @@ def extract_candidate_paths(bundle: dict, cfg: dict) -> list[str]:
     Paths outside the allowed prefixes/suffixes and any traversal, absolute,
     backslash, or NUL-bearing path are dropped. Never raises; returns [] when
     nothing matches.
+
+    A bundle carrying `sonarqube.issues` leads with the files those findings
+    name. Ranking console references is inference about where a failure came
+    from; a static-analysis finding is a direct statement of which file and
+    line are wrong, so guessing alongside it would only add noise.
     """
 
     try:
+        named = sonar_paths(bundle, cfg)
         references = _references(_search_texts(bundle), cfg)
         ranked = sorted(references, key=lambda path: _rank_key(path, references[path], cfg))
-        return ranked[: _max_candidates(cfg)]
+        ordered = named + [path for path in ranked if path not in named]
+        return ordered[: _max_candidates(cfg)]
+    except Exception:
+        return []
+
+
+def sonar_paths(bundle: dict, cfg: dict) -> list[str]:
+    """Return the writable files named by the bundle's SonarQube findings.
+
+    Inputs: an evidence bundle whose optional `sonarqube.issues` entries carry
+    a `path`, and the code cfg. Outputs: the distinct allowed paths in the
+    order the findings arrived. Never raises.
+    """
+
+    try:
+        sonar = bundle.get("sonarqube") if isinstance(bundle.get("sonarqube"), dict) else {}
+        issues = sonar.get("issues")
+        found: list[str] = []
+        for issue in issues if isinstance(issues, list) else []:
+            if not isinstance(issue, dict):
+                continue
+            path = str(issue.get("path") or "")
+            if path and path not in found and _is_allowed(path, cfg) and _is_safe(path):
+                found.append(path)
+        return found
     except Exception:
         return []
 

ariadne/services/hermes_code_defects.py

@@ -20,6 +20,7 @@ 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
@@ -190,6 +191,44 @@ def extract_assertion_defects(bundle: dict, cfg: dict) -> list[dict[str, Any]]:
         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."""
 
@@ -241,7 +280,12 @@ def _is_writable(path: str, cfg: dict) -> bool:
     return bool(prefixes) and any(path.startswith(p) for p in prefixes)
 
 
-ALL_CATEGORIES: tuple[str, ...] = (LINT_VIOLATION, UNDEFINED_NAME, FAILING_ASSERTION)
+ALL_CATEGORIES: tuple[str, ...] = (
+    SONARQUBE_ISSUE,
+    LINT_VIOLATION,
+    UNDEFINED_NAME,
+    FAILING_ASSERTION,
+)
 
 
 def enabled_categories(cfg: dict) -> tuple[str, ...]:
@@ -275,6 +319,8 @@ def extract_defects(bundle: dict, cfg: dict) -> list[dict[str, Any]]:
 
     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:
@@ -301,6 +347,12 @@ def defect_instruction(defects: list[dict[str, Any]]) -> str:
         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 "
@@ -313,6 +365,10 @@ 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:

ariadne/services/hermes_code_repair.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import base64
+import re
 from typing import Any
 
 import httpx
@@ -23,6 +24,7 @@ _OPEN_PULLS_LIMIT = 50
 _COMMIT_IDENTITY = {"name": "Hermes Agent", "email": "hermes@bstein.dev"}
 _COMMIT_OK_STATUSES = {HTTP_OK, HTTP_CREATED}
 _BRANCH_RETRY_STATUSES = {HTTP_NOT_FOUND, HTTP_UNPROCESSABLE}
+_BRANCH_UNSAFE = re.compile(r"[^A-Za-z0-9._-]+")
 
 
 def fetch_file(cfg: dict, path: str) -> tuple[str | None, str | None]:
@@ -82,18 +84,19 @@ def find_open_proposal(cfg: dict) -> dict[str, Any]:
 
 
 def push_branch(
-    cfg: dict, incident_id: str, build_number: int, patch: Any, patched_contents: str
+    cfg: dict, incident_id: str, ref: Any, patch: Any, patched_contents: str
 ) -> dict[str, Any]:
     """Commit the patched file to a fresh repair branch via the Gitea API.
 
-    Inputs: `cfg` as for `fetch_file`; the incident id, the failed build
-    number that names the branch, the validated ProposedPatch, and the fully
-    patched file contents. Outputs: {"branch", "committed", "error"}. Never
-    force-pushes, refuses master/main or empty branch names, never raises,
-    and never logs the token.
+    Inputs: `cfg` as for `fetch_file`; the incident id; `ref`, the token that
+    names the branch - the failed build number for a build-driven proposal, or
+    a sweep token when the proposal did not come from a build; the validated
+    ProposedPatch; and the fully patched file contents. Outputs:
+    {"branch", "committed", "error"}. Never force-pushes, refuses master/main
+    or empty branch names, never raises, and never logs the token.
     """
 
-    branch = _branch_name(build_number)
+    branch = _branch_name(ref)
     if not branch.strip() or branch in _PROTECTED_BRANCHES:
         return {"branch": branch, "committed": False, "error": f"refusing branch {branch!r}"}
     base_url = _base_url(cfg)
@@ -270,10 +273,16 @@ def _pr_body(incident_id: str, patch: Any, analysis: str) -> str:
     )
 
 
-def _branch_name(build_number: int) -> str:
-    """Derive the repair branch name for one failed build number."""
+def _branch_name(ref: Any) -> str:
+    """Derive the repair branch name from a build number or sweep token.
 
-    return f"hermes-repair/{build_number}"
+    Sanitized rather than interpolated: the token now reaches here from a
+    SonarQube finding key as well as from a build number, and a ref is one of
+    the few places where an unexpected character stops being cosmetic.
+    """
+
+    token = _BRANCH_UNSAFE.sub("-", str(ref)).strip("-")
+    return f"{_REPAIR_BRANCH_PREFIX}{token or 'unknown'}"
 
 
 def _json_payload(response: Any) -> dict[str, Any]:

ariadne/services/hermes_sonar_client.py

@@ -0,0 +1,193 @@
+"""Read open SonarQube findings for a project.
+
+Triage has so far been incident-driven: something fails, and the failure is
+what enters the flow. Static analysis is the opposite shape - a standing
+backlog that never fails a build and therefore never asks anyone for
+attention. On this instance that backlog is 139 open findings on Ariadne
+alone, each one already naming its file, its line, its rule and what is wrong.
+That is better-located evidence than the console text the code-repair flow
+usually works from, and it is being thrown away.
+
+This module only reads. It fetches findings and normalizes them into the same
+shape the patch flow already consumes; deciding which are worth acting on
+belongs to the sweep, and changing anything belongs to Ariadne.
+
+Two things are deliberately excluded rather than filtered later:
+
+Security hotspots are not findings. SonarQube models them as "needs a human to
+look at this", and the resolution is a review decision, not a code change. The
+quality gate on this instance fails on exactly that condition, and an
+automation that "fixed" it would be marking hotspots reviewed without review -
+defeating the control rather than satisfying it. They live behind a different
+endpoint, and this module does not call it.
+
+Findings already resolved or marked won't-fix carry a human's judgement. Asking
+a model to reopen that question wastes tokens and produces pull requests that
+argue with a decision someone already made.
+"""
+
+from __future__ import annotations
+
+import base64
+from typing import Any
+
+import httpx
+
+from ..utils.logging import get_logger
+
+
+logger = get_logger(__name__)
+
+HTTP_OK = 200
+
+CODE_SMELL = "CODE_SMELL"
+BUG = "BUG"
+VULNERABILITY = "VULNERABILITY"
+
+# Hotspots are absent on purpose; see the module docstring.
+ALL_ISSUE_TYPES: tuple[str, ...] = (CODE_SMELL, BUG, VULNERABILITY)
+
+_DEFAULT_TIMEOUT_SECONDS = 20.0
+_DEFAULT_PAGE_SIZE = 100
+_MAX_PAGE_SIZE = 500
+# Only findings nobody has ruled on. `resolved=false` already excludes fixed
+# ones; this excludes the ones a person decided to keep.
+_OPEN_STATUSES = "OPEN,CONFIRMED,REOPENED"
+
+
+def fetch_issues(cfg: dict, project: str) -> tuple[list[dict[str, Any]], str | None]:
+    """Fetch one project's open findings from SonarQube.
+
+    Inputs: `cfg` with sonar_base_url, sonar_token, optionally sonar_types,
+    sonar_severities and timeout_seconds; the SonarQube project key. Outputs:
+    (issues, error) with exactly one side meaningful - issues is [] on error.
+
+    Never raises and never logs the token. The token is sent as HTTP basic
+    with an empty password, which is how SonarQube accepts user tokens.
+    """
+
+    base_url = str(cfg.get("sonar_base_url") or "").rstrip("/")
+    if not base_url:
+        return [], "sonar base url is empty"
+    if not str(cfg.get("sonar_token") or ""):
+        return [], "sonar token is empty"
+    params = {
+        "componentKeys": project,
+        "resolved": "false",
+        "statuses": _OPEN_STATUSES,
+        "types": ",".join(_types(cfg)),
+        "ps": str(_page_size(cfg)),
+        "p": "1",
+    }
+    severities = _severities(cfg)
+    if severities:
+        params["severities"] = ",".join(severities)
+    try:
+        with httpx.Client(timeout=_timeout(cfg)) as client:
+            response = client.get(
+                f"{base_url}/api/issues/search", headers=_headers(cfg), params=params
+            )
+    except Exception as exc:
+        return [], f"sonar fetch failed: {exc}"
+    if response.status_code != HTTP_OK:
+        return [], f"sonar fetch http {response.status_code}"
+    try:
+        payload = response.json()
+    except Exception as exc:
+        return [], f"sonar response not json: {exc}"
+    return _normalized(payload), None
+
+
+def _normalized(payload: Any) -> list[dict[str, Any]]:
+    """Reduce the API payload to the fields the patch flow can act on."""
+
+    issues = payload.get("issues") if isinstance(payload, dict) else None
+    found: list[dict[str, Any]] = []
+    for issue in issues if isinstance(issues, list) else []:
+        if not isinstance(issue, dict):
+            continue
+        normalized = _normalize_issue(issue)
+        if normalized is not None:
+            found.append(normalized)
+    return found
+
+
+def _normalize_issue(issue: dict[str, Any]) -> dict[str, Any] | None:
+    """Normalize one finding, or None when it cannot be located in a file.
+
+    A finding with no line cannot be anchored to anything the patch validator
+    can check, so it is dropped here rather than sent to a model that would
+    have to guess where it belongs.
+    """
+
+    key = str(issue.get("key") or "").strip()
+    path = component_path(issue.get("component"))
+    line = issue.get("line")
+    if not key or not path or not isinstance(line, int) or isinstance(line, bool):
+        return None
+    return {
+        "key": key,
+        "path": path,
+        "line": line,
+        "rule": str(issue.get("rule") or "").strip(),
+        "severity": str(issue.get("severity") or "").strip(),
+        "type": str(issue.get("type") or "").strip(),
+        "effort": str(issue.get("effort") or "").strip(),
+        "message": " ".join(str(issue.get("message") or "").split())[:400],
+    }
+
+
+def component_path(component: Any) -> str:
+    """Return the repository-relative path from a SonarQube component key.
+
+    Inputs: a component key, which SonarQube formats as `project:path`.
+    Outputs: the path, or "" when the key names a project rather than a file.
+    """
+
+    text = str(component or "")
+    _, separator, path = text.partition(":")
+    return path.strip() if separator else ""
+
+
+def _types(cfg: dict) -> tuple[str, ...]:
+    """Return the finding types this deployment allows, defaulting to all."""
+
+    raw = cfg.get("sonar_types")
+    names = {str(item).strip().upper() for item in (raw or []) if str(item).strip()}
+    if not names:
+        return ALL_ISSUE_TYPES
+    return tuple(item for item in ALL_ISSUE_TYPES if item in names) or ALL_ISSUE_TYPES
+
+
+def _severities(cfg: dict) -> tuple[str, ...]:
+    """Return the severities this deployment restricts to, if any."""
+
+    raw = cfg.get("sonar_severities")
+    return tuple(str(item).strip().upper() for item in (raw or []) if str(item).strip())
+
+
+def _headers(cfg: dict) -> dict[str, str]:
+    """Build the basic-auth header SonarQube expects for a user token."""
+
+    token = str(cfg.get("sonar_token") or "")
+    encoded = base64.b64encode(f"{token}:".encode("utf-8")).decode("ascii")
+    return {"Authorization": f"Basic {encoded}", "Accept": "application/json"}
+
+
+def _page_size(cfg: dict) -> int:
+    """Clamp the page size to what the API accepts."""
+
+    try:
+        size = int(cfg.get("sonar_page_size") or _DEFAULT_PAGE_SIZE)
+    except (TypeError, ValueError):
+        return _DEFAULT_PAGE_SIZE
+    return max(1, min(size, _MAX_PAGE_SIZE))
+
+
+def _timeout(cfg: dict) -> float:
+    """Return the request timeout, defaulting when unset or unparseable."""
+
+    try:
+        return float(cfg.get("timeout_seconds") or _DEFAULT_TIMEOUT_SECONDS)
+    except (TypeError, ValueError):
+        return _DEFAULT_TIMEOUT_SECONDS

ariadne/services/hermes_sonar_sweep.py

@@ -0,0 +1,299 @@
+"""Turn standing SonarQube findings into reviewed pull requests.
+
+This is a second way into the same flow, not a second flow. Triage enters when
+a build fails; this enters on a schedule, because static analysis never fails
+anything and so never asks for attention on its own. From evidence collection
+onward both take exactly the same path: Hermes returns a patch as data, Ariadne
+validates it against the file it names, pushes a branch, and opens a pull
+request nobody merges automatically. The trust boundary is identical, and it
+has to be - a finding arriving from outside the build is not a reason to relax
+the gates that make a proposal safe to look at.
+
+What this adds over the build-driven path is precision. Console text has to be
+mined for a file and a line; a finding states both, plus the rule and what is
+wrong with it. That is why the sweep offers only the file a finding names
+rather than a ranked guess.
+
+Deliberately slow by default. `max_per_sweep` is one: a backlog of 139 findings
+could otherwise become 139 pull requests nobody reads, which would make the
+review gate theatre. One at a time, each one reviewed, is the pace at which the
+proposal is worth anything - and the open-proposal ceiling still applies on top,
+so the queue cannot grow while nobody is draining it.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from ..settings import settings
+from ..utils.logging import get_logger
+from . import hermes_code_flow, hermes_sonar_client
+
+
+logger = get_logger(__name__)
+
+SWEEP_EVENT_TYPE = "hermes_sonar_sweep"
+
+DEFAULT_MAX_PER_SWEEP = 1
+# A finding SonarQube estimates at hours of work is not a one-anchor patch. The
+# effort estimate is the cheapest available signal for "mechanical", so it is
+# the filter rather than a severity threshold: a trivial CRITICAL is a better
+# candidate than an involved MINOR.
+DEFAULT_MAX_EFFORT_MINUTES = 20
+
+_NO_PROJECTS = "no_projects_configured"
+
+
+def run_hermes_sonar_sweep(storage: Any) -> dict[str, Any]:
+    """Run one scheduled SonarQube sweep.
+
+    Inputs: a storage object providing record_event for the proposal event
+    log. Outputs: a summary dict for scheduler logging; {"status": "disabled"}
+    when the feature flag is off, which is the default.
+
+    Gated separately from triage on purpose. Triage reacts to a failure a
+    person already cares about; this opens pull requests nobody asked for, and
+    that is a decision an operator makes deliberately rather than inherits.
+    """
+
+    if not getattr(settings, "hermes_sonar_enabled", False):
+        return {"status": "disabled"}
+    hermes_cfg = {
+        "base_url": settings.hermes_api_url,
+        "api_key": settings.hermes_api_key,
+        "total_timeout_seconds": settings.hermes_run_timeout_seconds,
+    }
+    return sweep(storage, settings, hermes_cfg)
+
+
+def sweep(storage: Any, config: Any, hermes_cfg: dict) -> dict[str, Any]:
+    """Run one quality sweep across every configured SonarQube project.
+
+    Inputs: the incident event storage, a settings-like object exposing the
+    hermes_sonar_* and hermes_code_* fields, and the Hermes agent run config.
+    Outputs: {"proposed": n, "skipped": [...], "projects": n}.
+
+    Never raises: a sweep is a background improvement, and it must never be
+    able to take down the scheduler that also runs triage.
+    """
+
+    try:
+        return _sweep(storage, config, hermes_cfg)
+    except Exception as exc:
+        logger.info(
+            "hermes sonar sweep failed",
+            extra={"event": SWEEP_EVENT_TYPE, "status": "error", "detail": str(exc)},
+        )
+        return {"proposed": 0, "skipped": [f"sweep_failed: {exc}"], "projects": 0}
+
+
+def _sweep(storage: Any, config: Any, hermes_cfg: dict) -> dict[str, Any]:
+    """Fetch, select and propose for each configured project."""
+
+    projects = _projects(config)
+    if not projects:
+        return {"proposed": 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
+    skipped: list[str] = []
+    for project, job in sorted(projects.items()):
+        if proposed >= budget:
+            skipped.append(f"{project}: sweep budget spent")
+            break
+        outcome = _propose_for_project(
+            storage, project, job, sonar_cfg, code_cfg, hermes_cfg, config
+        )
+        proposed += outcome["proposed"]
+        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)}
+
+
+def _propose_for_project(  # noqa: PLR0913 - one project needs every config the flow does
+    storage: Any,
+    project: str,
+    job: str,
+    sonar_cfg: dict,
+    code_cfg: dict,
+    hermes_cfg: dict,
+    config: Any,
+) -> dict[str, Any]:
+    """Select one finding for a project and run the proposal flow for it."""
+
+    repo_cfg = hermes_code_flow.resolve_repo_config(job, code_cfg)
+    if repo_cfg is None:
+        return {"proposed": 0, "skipped": [f"{project}: job {job!r} maps to no repository"]}
+    issues, error = hermes_sonar_client.fetch_issues(sonar_cfg, project)
+    if error:
+        return {"proposed": 0, "skipped": [f"{project}: {error}"]}
+    chosen = select_issue(issues, repo_cfg, config)
+    if chosen is None:
+        return {"proposed": 0, "skipped": [f"{project}: no 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')}"]}
+
+
+def select_issue(issues: list[dict], repo_cfg: dict, config: Any) -> dict[str, Any] | None:
+    """Pick the single most mechanically fixable finding for one project.
+
+    Inputs: the normalized findings, the resolved per-repo cfg (whose write
+    allowlist decides what is patchable at all), and the settings object.
+    Outputs: one finding, or None when none qualify.
+
+    Ordered by effort, then severity, then key. Effort leads because it is the
+    closest available proxy for "one anchored change" - the shape the patch
+    validator can actually check - and the key breaks ties so a sweep that
+    proposes nothing today proposes the same finding tomorrow rather than
+    wandering the backlog.
+    """
+
+    ceiling = _max_effort_minutes(config)
+    eligible = [
+        issue
+        for issue in issues
+        if _is_writable(str(issue.get("path") or ""), repo_cfg)
+        and effort_minutes(issue.get("effort")) is not None
+        and (effort_minutes(issue.get("effort")) or 0) <= ceiling
+    ]
+    if not eligible:
+        return None
+    return sorted(eligible, key=_selection_key)[0]
+
+
+def _selection_key(issue: dict[str, Any]) -> tuple:
+    """Order findings by effort, then severity, then key for a stable pick."""
+
+    order = {"BLOCKER": 0, "CRITICAL": 1, "MAJOR": 2, "MINOR": 3, "INFO": 4}
+    return (
+        effort_minutes(issue.get("effort")) or 0,
+        order.get(str(issue.get("severity") or "").upper(), 9),
+        str(issue.get("key") or ""),
+    )
+
+
+def effort_minutes(effort: Any) -> int | None:
+    """Parse SonarQube's effort estimate into whole minutes.
+
+    Inputs: an effort string such as "5min", "1h30min" or "2d". Outputs: the
+    total in minutes, or None when it cannot be parsed - which the caller
+    treats as ineligible rather than as zero, since an unreadable estimate is
+    not evidence that the work is small.
+    """
+
+    text = str(effort or "").strip().lower()
+    if not text:
+        return None
+    total = 0
+    number = ""
+    units = {"d": 480, "h": 60, "min": 1}
+    index = 0
+    while index < len(text):
+        char = text[index]
+        if char.isdigit():
+            number += char
+            index += 1
+            continue
+        for suffix, factor in units.items():
+            if text.startswith(suffix, index) and number:
+                total += int(number) * factor
+                number = ""
+                index += len(suffix)
+                break
+        else:
+            return None
+    return total if not number and total else None
+
+
+def bundle_for(project: str, issue: dict[str, Any]) -> dict[str, Any]:
+    """Build the evidence bundle for one finding.
+
+    Inputs: the SonarQube project key and one normalized finding. Outputs: the
+    bundle the patch flow consumes, carrying an empty `jenkins` section so the
+    console-mining paths find nothing to guess from - the finding already says
+    which file and line are wrong.
+    """
+
+    return {
+        "incident_id": incident_id(project, issue),
+        "jenkins": {"console_failures": [], "console_tail": "", "failed_tests": []},
+        "sonarqube": {"project": project, "issues": [issue]},
+    }
+
+
+def incident_id(project: str, issue: dict[str, Any]) -> str:
+    """Name the incident for one finding, stable across sweeps."""
+
+    return f"sonar/{project}/{issue.get('key')}"
+
+
+def client_config(config: Any) -> dict[str, Any]:
+    """Build the SonarQube client cfg from a settings-like object."""
+
+    return {
+        "sonar_base_url": str(getattr(config, "hermes_sonar_url", "") or ""),
+        "sonar_token": str(getattr(config, "hermes_sonar_token", "") or ""),
+        "sonar_types": list(getattr(config, "hermes_sonar_types", None) or []),
+        "sonar_severities": list(getattr(config, "hermes_sonar_severities", None) or []),
+        "timeout_seconds": getattr(config, "hermes_sonar_timeout_seconds", None),
+    }
+
+
+def _projects(config: Any) -> dict[str, str]:
+    """Return the sonar-project -> jenkins-job map this deployment configured.
+
+    The job name is how a project reaches a repository: it is already mapped in
+    `hermes_code_repos`, so a sweep needs no second copy of that mapping and
+    cannot disagree with it. A project whose value is empty maps to itself.
+    """
+
+    raw = getattr(config, "hermes_sonar_projects", None)
+    if not isinstance(raw, dict):
+        return {}
+    return {str(key): str(value or key) for key, value in raw.items() if str(key).strip()}
+
+
+def _is_writable(path: str, cfg: dict) -> bool:
+    """Report whether a finding's file is one the patcher may write to."""
+
+    if not path:
+        return False
+    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 _max_per_sweep(config: Any) -> int:
+    """Return how many proposals one sweep may open."""
+
+    try:
+        value = int(getattr(config, "hermes_sonar_max_per_sweep", DEFAULT_MAX_PER_SWEEP))
+    except (TypeError, ValueError):
+        return DEFAULT_MAX_PER_SWEEP
+    return max(0, value)
+
+
+def _max_effort_minutes(config: Any) -> int:
+    """Return the effort ceiling above which a finding is left to a person."""
+
+    try:
+        value = int(getattr(config, "hermes_sonar_max_effort_minutes", DEFAULT_MAX_EFFORT_MINUTES))
+    except (TypeError, ValueError):
+        return DEFAULT_MAX_EFFORT_MINUTES
+    return max(1, value)

ariadne/settings.py

@@ -315,6 +315,16 @@ class Settings:
     jenkins_workspace_cleanup_cron: str
     testing_triage_cron: str
     hermes_autotriage_cron: str
+    hermes_sonar_cron: str
+    hermes_sonar_enabled: bool
+    hermes_sonar_url: str
+    hermes_sonar_token: str
+    hermes_sonar_projects: dict[str, str]
+    hermes_sonar_types: list[str]
+    hermes_sonar_severities: list[str]
+    hermes_sonar_max_per_sweep: int
+    hermes_sonar_max_effort_minutes: int
+    hermes_sonar_timeout_seconds: float
 
     opensearch_url: str
     opensearch_limit_bytes: int

ariadne/settings_hermes.py

@@ -51,6 +51,28 @@ def _hermes_autotriage_config() -> dict[str, Any]:
         ).rstrip("/"),
         "hermes_api_key": _env("ARIADNE_HERMES_API_KEY", ""),
         "hermes_run_timeout_seconds": _env_float("ARIADNE_HERMES_RUN_TIMEOUT_SECONDS", 420.0),
+        # SonarQube sweep. Disabled until projects are named: the map is
+        # sonar-project=jenkins-job, and the job already carries the repository
+        # mapping, so there is no second place for the two to disagree.
+        "hermes_sonar_enabled": _env_bool("ARIADNE_HERMES_SONAR_ENABLED", "false"),
+        "hermes_sonar_url": _env(
+            "ARIADNE_HERMES_SONAR_URL", "http://sonarqube.quality.svc.cluster.local:9000"
+        ).rstrip("/"),
+        "hermes_sonar_token": _env("ARIADNE_HERMES_SONAR_TOKEN", ""),
+        "hermes_sonar_projects": _pair_map(_env("ARIADNE_HERMES_SONAR_PROJECTS", "")),
+        "hermes_sonar_types": [
+            item.strip()
+            for item in _env("ARIADNE_HERMES_SONAR_TYPES", "CODE_SMELL,BUG").split(",")
+            if item.strip()
+        ],
+        "hermes_sonar_severities": [
+            item.strip()
+            for item in _env("ARIADNE_HERMES_SONAR_SEVERITIES", "").split(",")
+            if item.strip()
+        ],
+        "hermes_sonar_max_per_sweep": _env_int("ARIADNE_HERMES_SONAR_MAX_PER_SWEEP", 1),
+        "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"),
         "hermes_demo_fixture_configmap": _env("ARIADNE_HERMES_DEMO_FIXTURE_CONFIGMAP", "hermes-triage-demo-fixture"),
     }

ariadne/settings_sections.py

@@ -366,6 +366,8 @@ def _schedule_config() -> dict[str, Any]:
             "ARIADNE_SCHEDULE_HERMES_AUTOTRIAGE",
             "* * * * *",
         ),
+
+        "hermes_sonar_cron": _env("ARIADNE_HERMES_SONAR_CRON", "17 * * * *"),
         "wolf_oidc_cron": _env("ARIADNE_SCHEDULE_WOLF_OIDC", _env("ARIADNE_SCHEDULE_SUNSHINE_OIDC", "17 */6 * * *")),
     }
 

tests/test_hermes_code_candidates.py

@@ -276,3 +276,39 @@ def test_queue_imported_sources_only_follows_tests() -> None:
     queue, queued = ["tests/test_utils.py"], {"tests/test_utils.py"}
     module.queue_imported_sources(_CFG, "tests/test_utils.py", source, queue, queued)
     assert "ariadne/utils/errors.py" in queue
+
+
+def test_a_file_named_by_a_sonarqube_finding_leads_the_candidates() -> None:
+    """A finding states which file is wrong; console text only implies it."""
+
+    cfg = {"allowed_path_prefixes": ["ariadne/"], "allowed_suffixes": [".py"], "max_candidates": 3}
+    bundle = {
+        "jenkins": {"console_tail": "File \"ariadne/other.py\", line 3"},
+        "sonarqube": {"issues": [{"path": "ariadne/named.py"}]},
+    }
+
+    assert module.extract_candidate_paths(bundle, cfg)[0] == "ariadne/named.py"
+
+
+def test_sonarqube_paths_are_deduplicated_and_allowlisted() -> None:
+    cfg = {"allowed_path_prefixes": ["ariadne/"], "allowed_suffixes": [".py"]}
+    bundle = {
+        "sonarqube": {
+            "issues": [
+                {"path": "ariadne/a.py"},
+                {"path": "ariadne/a.py"},
+                {"path": "docs/readme.md"},
+                {"path": "../escape.py"},
+                {"path": ""},
+                "not-a-dict",
+            ]
+        }
+    }
+
+    assert module.sonar_paths(bundle, cfg) == ["ariadne/a.py"]
+
+
+def test_a_bundle_without_sonarqube_findings_names_no_paths() -> None:
+    assert module.sonar_paths({}, {}) == []
+    assert module.sonar_paths({"sonarqube": "nope"}, {}) == []
+    assert module.sonar_paths({"sonarqube": {"issues": None}}, {}) == []

tests/test_hermes_code_defects.py

@@ -2,6 +2,8 @@
 
 from __future__ import annotations
 
+import pytest
+
 from ariadne.services import hermes_code_defects as module
 
 
@@ -307,3 +309,85 @@ def test_an_unrecognised_category_is_ignored_not_trusted() -> None:
 def test_categories_keep_their_declared_order() -> None:
     cfg = {"fix_categories": [module.FAILING_ASSERTION, module.LINT_VIOLATION]}
     assert module.enabled_categories(cfg) == (module.LINT_VIOLATION, module.FAILING_ASSERTION)
+
+
+SONAR_CFG = {"allowed_path_prefixes": ["ariadne/"], "allowed_suffixes": [".py"]}
+
+
+def _sonar_bundle(*issues):
+    return {"sonarqube": {"project": "ariadne", "issues": list(issues)}}
+
+
+def test_a_sonarqube_finding_becomes_a_located_defect() -> None:
+    """The finding already names file, line and rule; nothing is inferred."""
+
+    bundle = _sonar_bundle(
+        {
+            "path": "ariadne/services/thing.py",
+            "line": 160,
+            "rule": "python:S1172",
+            "message": "Remove the unused function parameter.",
+        }
+    )
+
+    assert module.extract_sonar_defects(bundle, SONAR_CFG) == [
+        {
+            "category": module.SONARQUBE_ISSUE,
+            "tool": "sonarqube",
+            "path": "ariadne/services/thing.py",
+            "line": 160,
+            "rule": "python:S1172",
+            "message": "Remove the unused function parameter.",
+        }
+    ]
+
+
+@pytest.mark.parametrize(
+    "issue",
+    [
+        {"path": "docs/readme.md", "line": 1},
+        {"path": "", "line": 1},
+        "not-a-dict",
+    ],
+)
+def test_an_unusable_sonarqube_finding_is_dropped(issue) -> None:
+    assert module.extract_sonar_defects(_sonar_bundle(issue), SONAR_CFG) == []
+
+
+def test_a_bundle_without_findings_yields_nothing() -> None:
+    assert module.extract_sonar_defects({}, SONAR_CFG) == []
+    assert module.extract_sonar_defects({"sonarqube": "nope"}, SONAR_CFG) == []
+    assert module.extract_sonar_defects({"sonarqube": {"issues": None}}, SONAR_CFG) == []
+
+
+def test_sonarqube_findings_are_capped() -> None:
+    issues = [
+        {"path": f"ariadne/m{i}.py", "line": i, "rule": "r", "message": "m"} for i in range(40)
+    ]
+
+    assert len(module.extract_sonar_defects(_sonar_bundle(*issues), SONAR_CFG)) == 20
+
+
+def test_the_sonarqube_instruction_forbids_changing_behaviour() -> None:
+    """The build is green; a "fix" that alters behaviour is a regression."""
+
+    defects = module.extract_sonar_defects(
+        _sonar_bundle(
+            {"path": "ariadne/a.py", "line": 3, "rule": "python:S1172", "message": "Remove it."}
+        ),
+        SONAR_CFG,
+    )
+    instruction = module.defect_instruction(defects)
+
+    assert "ariadne/a.py line 3: python:S1172 Remove it." in instruction
+    assert "the build is not failing" in instruction
+    assert "Preserve the existing behaviour exactly" in instruction
+    assert "do not suppress the rule" in instruction
+
+
+def test_sonarqube_findings_can_be_disabled_by_category() -> None:
+    bundle = _sonar_bundle({"path": "ariadne/a.py", "line": 1, "rule": "r", "message": "m"})
+    cfg = {**SONAR_CFG, "fix_categories": ["lint_violation"]}
+
+    assert module.extract_defects(bundle, cfg) == []
+    assert module.extract_defects(bundle, SONAR_CFG)[0]["category"] == module.SONARQUBE_ISSUE

tests/test_hermes_sonar_client.py

@@ -0,0 +1,250 @@
+"""Tests for reading open SonarQube findings."""
+
+from __future__ import annotations
+
+import base64
+
+import httpx
+import pytest
+
+from ariadne.services import hermes_sonar_client as module
+
+
+CFG = {
+    "sonar_base_url": "http://sonarqube.quality.svc.cluster.local:9000/",
+    "sonar_token": "squ_token",
+    "timeout_seconds": 5,
+}
+
+ISSUE = {
+    "key": "AZ-1",
+    "component": "ariadne:ariadne/services/hermes_code_defects.py",
+    "line": 160,
+    "rule": "python:S1172",
+    "severity": "MAJOR",
+    "type": "CODE_SMELL",
+    "effort": "5min",
+    "message": 'Remove the unused function parameter "cfg".',
+}
+
+
+class _Response:
+    def __init__(self, status_code=200, payload=None, raises=False):
+        self.status_code = status_code
+        self._payload = payload if payload is not None else {"issues": [ISSUE]}
+        self._raises = raises
+
+    def json(self):
+        if self._raises:
+            raise ValueError("not json")
+        return self._payload
+
+
+class _Client:
+    """Stands in for httpx.Client, recording the single request made."""
+
+    last_url = ""
+    last_params: dict = {}
+    last_headers: dict = {}
+    response = _Response()
+    error: Exception | None = None
+
+    def __init__(self, *_args, **_kwargs):
+        pass
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, *_args):
+        return False
+
+    def get(self, url, headers=None, params=None):
+        _Client.last_url = url
+        _Client.last_params = params or {}
+        _Client.last_headers = headers or {}
+        if _Client.error is not None:
+            raise _Client.error
+        return _Client.response
+
+
+@pytest.fixture(autouse=True)
+def _stub_client(monkeypatch):
+    _Client.response = _Response()
+    _Client.error = None
+    monkeypatch.setattr(module.httpx, "Client", _Client)
+    return _Client
+
+
+def test_a_project_returns_normalized_findings() -> None:
+    issues, error = module.fetch_issues(CFG, "ariadne")
+
+    assert error is None
+    assert issues == [
+        {
+            "key": "AZ-1",
+            "path": "ariadne/services/hermes_code_defects.py",
+            "line": 160,
+            "rule": "python:S1172",
+            "severity": "MAJOR",
+            "type": "CODE_SMELL",
+            "effort": "5min",
+            "message": 'Remove the unused function parameter "cfg".',
+        }
+    ]
+
+
+def test_the_token_is_sent_as_basic_auth_with_an_empty_password() -> None:
+    """SonarQube accepts a user token only in that shape."""
+
+    module.fetch_issues(CFG, "ariadne")
+
+    expected = base64.b64encode(b"squ_token:").decode("ascii")
+    assert _Client.last_headers["Authorization"] == f"Basic {expected}"
+
+
+def test_only_unresolved_findings_nobody_has_ruled_on_are_requested() -> None:
+    """A won't-fix carries a human judgement; reopening it wastes tokens."""
+
+    module.fetch_issues(CFG, "ariadne")
+
+    assert _Client.last_params["resolved"] == "false"
+    assert _Client.last_params["statuses"] == "OPEN,CONFIRMED,REOPENED"
+    assert _Client.last_params["componentKeys"] == "ariadne"
+
+
+def test_security_hotspots_are_never_requested() -> None:
+    """Resolving a hotspot is a review decision, not a code change."""
+
+    module.fetch_issues(CFG, "ariadne")
+
+    assert "HOTSPOT" not in _Client.last_params["types"]
+    assert "SECURITY_HOTSPOT" not in _Client.last_params["types"]
+    assert set(_Client.last_params["types"].split(",")) == set(module.ALL_ISSUE_TYPES)
+
+
+def test_the_double_slash_in_the_url_is_avoided() -> None:
+    module.fetch_issues(CFG, "ariadne")
+
+    assert _Client.last_url == (
+        "http://sonarqube.quality.svc.cluster.local:9000/api/issues/search"
+    )
+
+
+def test_configured_types_and_severities_narrow_the_query() -> None:
+    module.fetch_issues(
+        {**CFG, "sonar_types": ["code_smell"], "sonar_severities": ["critical", "major"]},
+        "ariadne",
+    )
+
+    assert _Client.last_params["types"] == "CODE_SMELL"
+    assert _Client.last_params["severities"] == "CRITICAL,MAJOR"
+
+
+def test_unrecognised_types_fall_back_to_every_type() -> None:
+    module.fetch_issues({**CFG, "sonar_types": ["nonsense"]}, "ariadne")
+
+    assert set(_Client.last_params["types"].split(",")) == set(module.ALL_ISSUE_TYPES)
+
+
+@pytest.mark.parametrize(
+    ("size", "expected"),
+    [(None, "100"), (5, "5"), (9000, "500"), ("junk", "100")],
+)
+def test_the_page_size_is_clamped(size, expected) -> None:
+    module.fetch_issues({**CFG, "sonar_page_size": size}, "ariadne")
+
+    assert _Client.last_params["ps"] == expected
+
+
+def test_a_finding_with_no_line_is_dropped() -> None:
+    """Nothing can be anchored to a file-level finding, so it is not offered."""
+
+    _Client.response = _Response(payload={"issues": [{**ISSUE, "line": None}]})
+
+    issues, error = module.fetch_issues(CFG, "ariadne")
+
+    assert error is None
+    assert issues == []
+
+
+@pytest.mark.parametrize(
+    "issue",
+    [
+        {**ISSUE, "key": ""},
+        {**ISSUE, "component": "ariadne"},
+        {**ISSUE, "line": True},
+        "not-a-dict",
+    ],
+)
+def test_an_unusable_finding_is_dropped(issue) -> None:
+    _Client.response = _Response(payload={"issues": [issue]})
+
+    assert module.fetch_issues(CFG, "ariadne")[0] == []
+
+
+def test_a_long_message_is_clipped() -> None:
+    _Client.response = _Response(payload={"issues": [{**ISSUE, "message": "x " * 500}]})
+
+    issues, _ = module.fetch_issues(CFG, "ariadne")
+
+    assert len(issues[0]["message"]) <= 400
+
+
+def test_a_missing_base_url_makes_no_request() -> None:
+    issues, error = module.fetch_issues({**CFG, "sonar_base_url": ""}, "ariadne")
+
+    assert issues == []
+    assert error == "sonar base url is empty"
+
+
+def test_a_missing_token_makes_no_request() -> None:
+    issues, error = module.fetch_issues({**CFG, "sonar_token": ""}, "ariadne")
+
+    assert issues == []
+    assert error == "sonar token is empty"
+
+
+def test_a_transport_failure_is_reported_not_raised() -> None:
+    _Client.error = httpx.ConnectError("refused")
+
+    issues, error = module.fetch_issues(CFG, "ariadne")
+
+    assert issues == []
+    assert "sonar fetch failed" in error
+
+
+def test_a_non_200_is_reported() -> None:
+    _Client.response = _Response(status_code=503)
+
+    issues, error = module.fetch_issues(CFG, "ariadne")
+
+    assert issues == []
+    assert error == "sonar fetch http 503"
+
+
+def test_an_unparseable_body_is_reported() -> None:
+    _Client.response = _Response(raises=True)
+
+    issues, error = module.fetch_issues(CFG, "ariadne")
+
+    assert issues == []
+    assert "sonar response not json" in error
+
+
+def test_a_payload_that_is_not_a_dict_yields_nothing() -> None:
+    _Client.response = _Response(payload=["nope"])
+
+    assert module.fetch_issues(CFG, "ariadne") == ([], None)
+
+
+def test_the_timeout_falls_back_when_unparseable() -> None:
+    assert module._timeout({"timeout_seconds": "soon"}) == module._DEFAULT_TIMEOUT_SECONDS
+    assert module._timeout({}) == module._DEFAULT_TIMEOUT_SECONDS
+
+
+@pytest.mark.parametrize(
+    ("component", "expected"),
+    [("proj:src/a.py", "src/a.py"), ("proj", ""), (None, ""), ("proj: a.py ", "a.py")],
+)
+def test_the_component_key_yields_the_repository_path(component, expected) -> None:
+    assert module.component_path(component) == expected

tests/test_hermes_sonar_sweep.py

@@ -0,0 +1,309 @@
+"""Tests for the scheduled SonarQube quality sweep."""
+
+from __future__ import annotations
+
+from types import SimpleNamespace
+
+import pytest
+
+from ariadne.services import hermes_sonar_sweep as module
+
+
+REPO_CFG = {
+    "owner": "bstein",
+    "repo": "ariadne",
+    "allowed_path_prefixes": ["ariadne/"],
+    "allowed_suffixes": [".py"],
+}
+
+
+def _issue(**overrides):
+    issue = {
+        "key": "AZ-1",
+        "path": "ariadne/services/hermes_code_defects.py",
+        "line": 160,
+        "rule": "python:S1172",
+        "severity": "MAJOR",
+        "type": "CODE_SMELL",
+        "effort": "5min",
+        "message": "Remove the unused function parameter.",
+    }
+    issue.update(overrides)
+    return issue
+
+
+def _config(**overrides):
+    values = {
+        "hermes_sonar_projects": {"ariadne": "ariadne"},
+        "hermes_sonar_url": "http://sonar:9000",
+        "hermes_sonar_token": "t",
+        "hermes_sonar_types": ["CODE_SMELL"],
+        "hermes_sonar_severities": [],
+        "hermes_sonar_max_per_sweep": 1,
+        "hermes_sonar_max_effort_minutes": 20,
+        "hermes_sonar_timeout_seconds": 5,
+    }
+    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))
+
+
+@pytest.fixture
+def wiring(monkeypatch):
+    """Stub the repo resolution, the client and the proposal flow."""
+
+    calls = {"proposals": [], "issues": [_issue()], "error": None, "result": {"status": "pr_opened"}}
+    monkeypatch.setattr(module.hermes_code_flow, "code_config", lambda config: {})
+    monkeypatch.setattr(module.hermes_code_flow, "resolve_repo_config", lambda job, cfg: REPO_CFG)
+    monkeypatch.setattr(
+        module.hermes_sonar_client, "fetch_issues",
+        lambda cfg, project: (calls["issues"], calls["error"]),
+    )
+
+    def _propose(storage, **kwargs):
+        calls["proposals"].append(kwargs)
+        return calls["result"]
+
+    monkeypatch.setattr(module.hermes_code_flow, "propose_code_fix", _propose)
+    return calls
+
+
+@pytest.mark.parametrize(
+    ("effort", "expected"),
+    [
+        ("5min", 5),
+        ("1h", 60),
+        ("1h30min", 90),
+        ("2d", 960),
+        ("", None),
+        ("soon", None),
+        ("30", None),
+        (None, None),
+    ],
+)
+def test_the_effort_estimate_is_parsed_into_minutes(effort, expected) -> None:
+    assert module.effort_minutes(effort) == expected
+
+
+def test_the_cheapest_finding_is_chosen_first() -> None:
+    """Effort is the closest proxy for a change the validator can check."""
+
+    issues = [
+        _issue(key="a", effort="15min"),
+        _issue(key="b", effort="5min"),
+        _issue(key="c", effort="10min"),
+    ]
+
+    assert module.select_issue(issues, REPO_CFG, _config())["key"] == "b"
+
+
+def test_severity_breaks_a_tie_on_effort() -> None:
+    issues = [_issue(key="a", severity="MINOR"), _issue(key="b", severity="CRITICAL")]
+
+    assert module.select_issue(issues, REPO_CFG, _config())["key"] == "b"
+
+
+def test_the_key_breaks_a_full_tie_so_the_pick_is_stable() -> None:
+    """A sweep that proposes nothing today must retry the same finding."""
+
+    issues = [_issue(key="zz"), _issue(key="aa")]
+
+    assert module.select_issue(issues, REPO_CFG, _config())["key"] == "aa"
+    assert module.select_issue(list(reversed(issues)), REPO_CFG, _config())["key"] == "aa"
+
+
+def test_an_expensive_finding_is_left_to_a_person() -> None:
+    issues = [_issue(effort="4h")]
+
+    assert module.select_issue(issues, REPO_CFG, _config()) is None
+
+
+def test_an_unparseable_effort_is_not_treated_as_free() -> None:
+    """An unreadable estimate is not evidence that the work is small."""
+
+    assert module.select_issue([_issue(effort="dunno")], REPO_CFG, _config()) is None
+
+
+@pytest.mark.parametrize(
+    "path",
+    ["docs/readme.md", "ariadne/thing.txt", "scripts/tool.py", ""],
+)
+def test_a_finding_outside_the_write_allowlist_is_never_chosen(path) -> None:
+    assert module.select_issue([_issue(path=path)], REPO_CFG, _config()) is None
+
+
+def test_a_sweep_opens_one_proposal_for_the_chosen_finding(wiring) -> None:
+    storage = _Storage()
+
+    result = module.sweep(storage, _config(), {})
+
+    assert result == {"proposed": 1, "skipped": [], "projects": 1}
+    proposal = wiring["proposals"][0]
+    assert proposal["incident_id"] == "sonar/ariadne/AZ-1"
+    assert proposal["job"] == "ariadne"
+    assert proposal["build_number"] == "sonar-AZ-1"
+
+
+def test_the_bundle_offers_only_the_file_the_finding_names() -> None:
+    """The finding states the file; ranking console text alongside adds noise."""
+
+    bundle = module.bundle_for("ariadne", _issue())
+
+    assert bundle["sonarqube"] == {"project": "ariadne", "issues": [_issue()]}
+    assert bundle["jenkins"]["console_failures"] == []
+    assert bundle["jenkins"]["console_tail"] == ""
+    assert bundle["incident_id"] == "sonar/ariadne/AZ-1"
+
+
+def test_the_sweep_budget_stops_after_its_quota(wiring) -> None:
+    """A backlog must not become a pull request queue nobody drains."""
+
+    config = _config(hermes_sonar_projects={"a": "a", "b": "b", "c": "c"})
+
+    result = module.sweep(_Storage(), config, {})
+
+    assert result["proposed"] == 1
+    assert len(wiring["proposals"]) == 1
+    assert any("budget spent" in item for item in result["skipped"])
+
+
+def test_a_declined_proposal_is_reported_and_does_not_spend_budget(wiring) -> None:
+    wiring["result"] = {"status": "human_required", "reason": "open_proposal_limit_reached"}
+    config = _config(hermes_sonar_projects={"a": "a", "b": "b"})
+
+    result = module.sweep(_Storage(), config, {})
+
+    assert result["proposed"] == 0
+    assert len(wiring["proposals"]) == 2
+    assert "a: open_proposal_limit_reached" 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 wiring["proposals"] == []
+
+
+def test_a_project_naming_no_repository_is_skipped(monkeypatch, wiring) -> None:
+    monkeypatch.setattr(module.hermes_code_flow, "resolve_repo_config", lambda job, cfg: None)
+
+    result = module.sweep(_Storage(), _config(), {})
+
+    assert result["proposed"] == 0
+    assert "maps to no repository" in result["skipped"][0]
+    assert wiring["proposals"] == []
+
+
+def test_a_fetch_error_is_reported_not_raised(wiring) -> None:
+    wiring["error"] = "sonar fetch http 503"
+
+    result = module.sweep(_Storage(), _config(), {})
+
+    assert result["proposed"] == 0
+    assert result["skipped"] == ["ariadne: sonar fetch http 503"]
+
+
+def test_a_project_with_nothing_mechanical_is_reported(wiring) -> None:
+    wiring["issues"] = [_issue(effort="8h")]
+
+    result = module.sweep(_Storage(), _config(), {})
+
+    assert result["skipped"] == ["ariadne: no mechanically fixable finding"]
+
+
+def test_a_sweep_never_takes_down_the_scheduler(monkeypatch) -> None:
+    """Triage shares this scheduler; a background sweep must not stop it."""
+
+    monkeypatch.setattr(
+        module.hermes_code_flow, "code_config",
+        lambda config: (_ for _ in ()).throw(RuntimeError("boom")),
+    )
+
+    result = module.sweep(_Storage(), _config(), {})
+
+    assert result["proposed"] == 0
+    assert "sweep_failed" in result["skipped"][0]
+
+
+def test_a_project_with_no_job_maps_to_itself() -> None:
+    assert module._projects(_config(hermes_sonar_projects={"ariadne": ""})) == {
+        "ariadne": "ariadne"
+    }
+    assert module._projects(SimpleNamespace(hermes_sonar_projects="nope")) == {}
+
+
+def test_the_client_config_is_built_from_settings() -> None:
+    cfg = module.client_config(_config())
+
+    assert cfg["sonar_base_url"] == "http://sonar:9000"
+    assert cfg["sonar_token"] == "t"
+    assert cfg["sonar_types"] == ["CODE_SMELL"]
+
+
+@pytest.mark.parametrize(
+    ("attr", "func", "bad", "default"),
+    [
+        ("hermes_sonar_max_per_sweep", "_max_per_sweep", "junk", module.DEFAULT_MAX_PER_SWEEP),
+        (
+            "hermes_sonar_max_effort_minutes",
+            "_max_effort_minutes",
+            "junk",
+            module.DEFAULT_MAX_EFFORT_MINUTES,
+        ),
+    ],
+)
+def test_an_unparseable_bound_falls_back_to_its_default(attr, func, bad, default) -> None:
+    assert getattr(module, func)(_config(**{attr: bad})) == default
+
+
+def test_the_bounds_are_floored() -> None:
+    assert module._max_per_sweep(_config(hermes_sonar_max_per_sweep=-4)) == 0
+    assert module._max_effort_minutes(_config(hermes_sonar_max_effort_minutes=0)) == 1
+
+
+def test_the_sweep_is_disabled_until_an_operator_turns_it_on(monkeypatch) -> None:
+    """It opens pull requests nobody asked for; that is a deliberate choice."""
+
+    monkeypatch.setattr(module, "settings", SimpleNamespace(hermes_sonar_enabled=False))
+
+    assert module.run_hermes_sonar_sweep(_Storage()) == {"status": "disabled"}
+
+
+def test_the_default_deployment_leaves_it_off() -> None:
+    """A fresh deployment must not start opening pull requests on its own."""
+
+    from ariadne.settings import settings as real_settings
+
+    assert real_settings.hermes_sonar_enabled is False
+
+
+def test_the_scheduled_entry_point_runs_the_sweep_when_enabled(monkeypatch) -> None:
+    seen = {}
+    monkeypatch.setattr(
+        module,
+        "settings",
+        SimpleNamespace(
+            hermes_sonar_enabled=True,
+            hermes_api_url="http://hermes:8642",
+            hermes_api_key="k",
+            hermes_run_timeout_seconds=420.0,
+        ),
+    )
+    monkeypatch.setattr(
+        module, "sweep", lambda storage, config, cfg: seen.update(cfg=cfg) or {"proposed": 0}
+    )
+
+    assert module.run_hermes_sonar_sweep(_Storage()) == {"proposed": 0}
+    assert seen["cfg"] == {
+        "base_url": "http://hermes:8642",
+        "api_key": "k",
+        "total_timeout_seconds": 420.0,
+    }