Compare commits

..

1 Commits

24 changed files with 53 additions and 1495 deletions

View File

@ -217,27 +217,6 @@ def _register_admin_routes(app: FastAPI, require_auth: Callable, deps: Callable[
module = deps() module = deps()
return JSONResponse(module.run_testing_triage_diagnosis(module.storage)) return JSONResponse(module.run_testing_triage_diagnosis(module.storage))
# The scheduler runs this every minute, which is cron's finest granularity
# and still up to sixty seconds of silence after a build goes red. That is
# the single largest delay between a failure happening and the system
# visibly reacting to it. Running the same tick on demand collapses it;
# the tick is idempotent - incidents dedupe on job and build number - so
# an extra run can only ever be a no-op.
@app.post("/api/admin/hermes/autotriage/run")
def run_hermes_autotriage_now(ctx: AuthContext = Depends(require_auth)) -> JSONResponse:
"""Run one Hermes auto-triage tick immediately."""
module = deps()
module._require_admin(ctx)
return JSONResponse(module.run_hermes_autotriage(module.storage))
@app.post("/api/internal/hermes/autotriage/run")
def run_hermes_autotriage_now_internal() -> JSONResponse:
"""Run one Hermes auto-triage tick for trusted internal callers."""
module = deps()
return JSONResponse(module.run_hermes_autotriage(module.storage))
@app.post("/api/admin/access/requests/{username}/approve") @app.post("/api/admin/access/requests/{username}/approve")
async def approve_access_request( async def approve_access_request(
username: str, username: str,

View File

@ -2,7 +2,7 @@ from __future__ import annotations
from typing import Any from typing import Any
from .cluster_state_contract import * from .cluster_state_contract import ClusterStateSummary, SignalContext, _NODE_WORKLOAD_TOP, _PROFILE_LIMIT
ProfileRows = list[dict[str, Any]] ProfileRows = list[dict[str, Any]]
NodeWorkloadMap = dict[str, dict[str, int]] NodeWorkloadMap = dict[str, dict[str, int]]

View File

@ -1,10 +1,9 @@
from __future__ import annotations from __future__ import annotations
from dataclasses import dataclass, field from dataclasses import dataclass
import json import json
from typing import Any from typing import Any
from ariadne.services import hermes_code_suggestion as code_suggestion_field
from ariadne.services import hermes_suggested_remediation as suggestion_field from ariadne.services import hermes_suggested_remediation as suggestion_field
@ -25,10 +24,7 @@ _TOP_LEVEL_KEYS = {
} }
# Optional because the field only applies when nothing in the allowlist fit, and # Optional because the field only applies when nothing in the allowlist fit, and
# because a response written before it existed must keep validating unchanged. # because a response written before it existed must keep validating unchanged.
_OPTIONAL_TOP_LEVEL_KEYS = { _OPTIONAL_TOP_LEVEL_KEYS = {suggestion_field.SUGGESTION_KEY}
suggestion_field.SUGGESTION_KEY,
code_suggestion_field.SUGGESTIONS_KEY,
}
_STRING_FIELDS = ("incident_id", "classification", "first_failed_gate", "reason") _STRING_FIELDS = ("incident_id", "classification", "first_failed_gate", "reason")
_FACT_KEYS = {"statement", "source", "reference"} _FACT_KEYS = {"statement", "source", "reference"}
_ACTION_KEYS = {"type", "id"} _ACTION_KEYS = {"type", "id"}
@ -83,7 +79,6 @@ class TriageDecision:
human_required: bool human_required: bool
reason: str reason: str
suggested_remediation: suggestion_field.SuggestedRemediation | None = None suggested_remediation: suggestion_field.SuggestedRemediation | None = None
code_suggestions: list[code_suggestion_field.CodeSuggestion] = field(default_factory=list)
@dataclass(frozen=True) @dataclass(frozen=True)
@ -262,7 +257,6 @@ def _validate_payload(payload: dict[str, Any], expected_incident_id: str) -> str
or _validate_facts(payload["facts"]) or _validate_facts(payload["facts"])
or _validate_requested_action(payload["requested_action"]) or _validate_requested_action(payload["requested_action"])
or suggestion_field.validate(payload) or suggestion_field.validate(payload)
or code_suggestion_field.validate(payload)
) )
if type_error: if type_error:
return type_error return type_error
@ -335,7 +329,6 @@ def _decision_from_payload(payload: dict[str, Any]) -> TriageDecision:
human_required=payload["human_required"], human_required=payload["human_required"],
reason=payload["reason"], reason=payload["reason"],
suggested_remediation=suggestion_field.from_payload(payload), suggested_remediation=suggestion_field.from_payload(payload),
code_suggestions=code_suggestion_field.from_payload(payload),
) )

View File

@ -12,7 +12,6 @@ from dataclasses import dataclass
import json import json
from typing import Any from typing import Any
from . import hermes_code_suggestion as code_suggestion_field
from . import hermes_suggested_remediation as suggestion_field from . import hermes_suggested_remediation as suggestion_field
from .hermes_autotriage_metrics import ( from .hermes_autotriage_metrics import (
HERMES_TRIAGE_ACTION_TOTAL, HERMES_TRIAGE_ACTION_TOTAL,
@ -214,9 +213,6 @@ def outcome_phase(outcome: Any) -> dict[str, Any]:
"suggested_remediation": suggestion_field.as_detail( "suggested_remediation": suggestion_field.as_detail(
getattr(decision, "suggested_remediation", None) getattr(decision, "suggested_remediation", None)
), ),
"code_suggestions": code_suggestion_field.as_detail(
getattr(decision, "code_suggestions", None)
),
} }

View File

@ -7,7 +7,6 @@ from typing import Any
import httpx import httpx
from ..utils.logging import get_logger from ..utils.logging import get_logger
from . import hermes_sonar_client
logger = get_logger(__name__) logger = get_logger(__name__)
@ -24,11 +23,6 @@ _REPAIR_BRANCH_PREFIX = "hermes-repair/"
_OPEN_PULLS_LIMIT = 50 _OPEN_PULLS_LIMIT = 50
_COMMIT_IDENTITY = {"name": "Hermes Agent", "email": "hermes@bstein.dev"} _COMMIT_IDENTITY = {"name": "Hermes Agent", "email": "hermes@bstein.dev"}
_COMMIT_OK_STATUSES = {HTTP_OK, HTTP_CREATED} _COMMIT_OK_STATUSES = {HTTP_OK, HTTP_CREATED}
# The Hermes console reopens a finished run from its id at this route.
RUN_PATH = "/chat?resume="
# Every proposal title is built from this, so the incident an open pull
# request belongs to can be read back without storing a second index.
PR_TITLE_PREFIX = "fix(hermes): repair "
_BRANCH_RETRY_STATUSES = {HTTP_NOT_FOUND, HTTP_UNPROCESSABLE} _BRANCH_RETRY_STATUSES = {HTTP_NOT_FOUND, HTTP_UNPROCESSABLE}
_BRANCH_UNSAFE = re.compile(r"[^A-Za-z0-9._-]+") _BRANCH_UNSAFE = re.compile(r"[^A-Za-z0-9._-]+")
@ -89,45 +83,6 @@ def find_open_proposal(cfg: dict) -> dict[str, Any]:
return _oldest_repair_pull(response, _base_branch(cfg)) return _oldest_repair_pull(response, _base_branch(cfg))
def open_proposal_incidents(cfg: dict) -> tuple[list[str], str | None]:
"""List the incidents that already have an open repair pull request.
Inputs: `cfg` as for `fetch_file`. Outputs: (incident_ids, error).
Read back from the pull request titles rather than from a stored index,
because the pull requests are the thing that actually exists - an index
could disagree with them, and the whole point of the check is to not
propose something a reviewer is already looking at.
Fails open: on any error the list is empty, so a lookup failure produces a
duplicate rather than silently suppressing real work.
"""
base_url = _base_url(cfg)
if not base_url:
return [], "gitea base url is empty"
try:
with httpx.Client(timeout=_timeout(cfg)) as client:
response = client.get(
f"{base_url}/api/v1/repos/{_owner(cfg)}/{_repo(cfg)}/pulls",
headers=_headers(cfg),
params={"state": "open", "limit": _OPEN_PULLS_LIMIT},
)
if response.status_code != HTTP_OK:
return [], f"open proposal lookup http {response.status_code}"
payload = response.json()
except Exception as exc:
return [], f"open proposal lookup failed: {exc}"
if not isinstance(payload, list):
return [], "open proposal payload is not a list"
found = []
for pull in payload:
title = str(pull.get("title") or "") if isinstance(pull, dict) else ""
if title.startswith(PR_TITLE_PREFIX):
found.append(title[len(PR_TITLE_PREFIX) :].strip())
return found, None
def push_branch( def push_branch(
cfg: dict, incident_id: str, ref: Any, patch: Any, patched_contents: str cfg: dict, incident_id: str, ref: Any, patch: Any, patched_contents: str
) -> dict[str, Any]: ) -> dict[str, Any]:
@ -179,7 +134,7 @@ def open_pull_request( # noqa: PLR0913 - the body needs the full proposal prove
payload = { payload = {
"head": branch, "head": branch,
"base": _base_branch(cfg), "base": _base_branch(cfg),
"title": f"{PR_TITLE_PREFIX}{incident_id}", "title": f"fix(hermes): repair {incident_id}",
"body": _pr_body(incident_id, patch, analysis, run_id, cfg), "body": _pr_body(incident_id, patch, analysis, run_id, cfg),
} }
try: try:
@ -301,19 +256,6 @@ def _pr_result(response: Any) -> dict[str, Any]:
return {"pr_number": number, "url": url, "error": None} return {"pr_number": number, "url": url, "error": None}
def run_url(ui_url: str, run_id: str) -> str:
"""Build the deep link that reopens one Hermes run in its console.
Inputs: the Hermes UI base url and a run id. Outputs: the resume link, or
"" when either is missing. Kept here so the pull request and the demo
monitor cannot drift into pointing at different pages.
"""
base = str(ui_url or "").rstrip("/")
run = str(run_id or "").strip()
return f"{base}{RUN_PATH}{run}" if base and run else ""
def _pr_body(incident_id: str, patch: Any, analysis: str, run_id: str, cfg: dict) -> str: def _pr_body(incident_id: str, patch: Any, analysis: str, run_id: str, cfg: dict) -> str:
"""Render the markdown pull-request body for human review. """Render the markdown pull-request body for human review.
@ -329,26 +271,16 @@ def _pr_body(incident_id: str, patch: Any, analysis: str, run_id: str, cfg: dict
"", "",
f"**Incident:** {incident_id}", f"**Incident:** {incident_id}",
f"**File:** `{patch.path}`", f"**File:** `{patch.path}`",
]
finding = hermes_sonar_client.issue_url(incident_id, cfg.get("sonar_ui_url"))
if finding:
# A sweep proposal exists because of one finding; the reviewer's first
# question is what it said, and an issue key is not an answer.
lines.append(f"**SonarQube finding:** {finding}")
lines += [
f"**Analysis:** {analysis}", f"**Analysis:** {analysis}",
f"**Rationale:** {patch.rationale}", f"**Rationale:** {patch.rationale}",
] ]
if run_id: if run_id:
lines.append(f"**Hermes run:** `{run_id}`")
ui_url = str(cfg.get("hermes_ui_url") or "").rstrip("/") ui_url = str(cfg.get("hermes_ui_url") or "").rstrip("/")
if ui_url: if ui_url:
lines.append(f"**Hermes run:** [{run_id}]({run_url(ui_url, run_id)})")
lines.append( lines.append(
"That link opens the run itself: the prompt Hermes was given, the evidence " f"The prompt this run was given and every tool call it made are at {ui_url}."
"bundle it read, the tools it called, and the JSON it returned."
) )
else:
lines.append(f"**Hermes run:** `{run_id}`")
lines += [ lines += [
"", "",
"Proposed by Hermes; validated and pushed by Ariadne; " "Proposed by Hermes; validated and pushed by Ariadne; "

View File

@ -55,8 +55,6 @@ def build_config(config: Any) -> dict[str, Any]:
"repos": _parse_repo_map(getattr(config, "hermes_code_repos", "")), "repos": _parse_repo_map(getattr(config, "hermes_code_repos", "")),
# Shown in the pull request so a reviewer can read the run that wrote it. # Shown in the pull request so a reviewer can read the run that wrote it.
"hermes_ui_url": str(getattr(config, "hermes_ui_url", "") or ""), "hermes_ui_url": str(getattr(config, "hermes_ui_url", "") or ""),
# Shown in sweep proposals so the finding itself is one click away.
"sonar_ui_url": str(getattr(config, "hermes_sonar_ui_url", "") or ""),
"job_prefixes": _parse_list_map(getattr(config, "hermes_code_prefixes", "")), "job_prefixes": _parse_list_map(getattr(config, "hermes_code_prefixes", "")),
"job_suffixes": _parse_list_map(getattr(config, "hermes_code_suffixes", "")), "job_suffixes": _parse_list_map(getattr(config, "hermes_code_suffixes", "")),
"job_base_branches": dict(_parse_pairs(getattr(config, "hermes_code_base_branches", ""))), "job_base_branches": dict(_parse_pairs(getattr(config, "hermes_code_base_branches", ""))),

View File

@ -1,157 +0,0 @@
"""Let an escalated diagnosis carry the fix it could not apply.
Ariadne opens a pull request when it can: the repository is mapped, the file
is inside the write allowlist, and the change is one anchored snippet its
validator can check. When any of that fails, the incident escalates to a human
with a diagnosis and nothing else - even though the model that wrote the
diagnosis frequently knows exactly what the fix is. That knowledge was being
discarded at the moment it was most useful, because the cases where no patch
is possible are precisely the cases a maintainer has to do by hand.
So a diagnosis may carry code suggestions: the file, what to change, and the
code to change it to. This is not a patch and is deliberately not treated as
one. Nothing here is validated against the file, anchored, applied, or pushed;
it is rendered into the issue as a starting point for a person who will read
it, judge it, and write the real change themselves.
That distinction is the whole safety story. A patch has to survive the
validator because Ariadne acts on it. A suggestion is read by a human who is
already going to edit this file, so being wrong costs them a moment's thought
rather than a bad commit - which is also why suggestions are allowed to
describe changes too large or too diffuse for the patcher to have attempted.
"""
from __future__ import annotations
from dataclasses import dataclass
from typing import Any
SUGGESTIONS_KEY = "code_suggestions"
_KEYS = {"path", "explanation", "code"}
_MAX_SUGGESTIONS = 3
_MAX_EXPLANATION = 600
_MAX_CODE = 1600
_FENCE = "```"
_FENCE_ESCAPE = "```"
_TRUNCATED = "\n... (truncated)"
@dataclass(frozen=True)
class CodeSuggestion:
"""Represent one suggested change a person is expected to review.
Inputs: a validated `code_suggestions[]` entry. Outputs: the file it
concerns, why the change is needed, and the suggested code.
Never applied and never validated against the file. It is advice.
"""
path: str
explanation: str
code: str
def validate(payload: dict[str, Any]) -> str | None:
"""Validate the optional code_suggestions field of a diagnosis.
Inputs: the whole parsed response payload. Outputs: a specific reject
reason, or None when the field is absent, null, or well-formed.
Bounded at three. A diagnosis that suggests a dozen changes has stopped
diagnosing and started rewriting, and an issue that long buries the reason
a person was called in the first place.
"""
suggestions = payload.get(SUGGESTIONS_KEY)
if suggestions is None:
return None
if not isinstance(suggestions, list):
return "code_suggestions_invalid: must be a list or null"
if len(suggestions) > _MAX_SUGGESTIONS:
return f"code_suggestions_invalid: at most {_MAX_SUGGESTIONS} suggestions"
for index, suggestion in enumerate(suggestions):
error = _validate_one(suggestion, index)
if error:
return error
return None
def _validate_one(suggestion: Any, index: int) -> str | None:
"""Validate one suggestion entry's shape and fields."""
if not isinstance(suggestion, dict):
return f"code_suggestions_invalid: [{index}] must be an object"
if set(suggestion) != _KEYS:
return f"code_suggestions_invalid: [{index}] must have exactly path, explanation, code"
for key in sorted(_KEYS):
if not isinstance(suggestion[key], str) or not suggestion[key].strip():
return f"code_suggestions_invalid: [{index}] {key} must be a non-empty string"
return None
def from_payload(payload: dict[str, Any]) -> list[CodeSuggestion]:
"""Build the typed suggestions from an already-validated payload.
Inputs: the parsed response payload. Outputs: the suggestions, clipped so
a long one cannot crowd the diagnosis out of a bounded issue body.
"""
suggestions = payload.get(SUGGESTIONS_KEY)
if not isinstance(suggestions, list):
return []
built = []
for suggestion in suggestions[:_MAX_SUGGESTIONS]:
if not isinstance(suggestion, dict):
continue
built.append(
CodeSuggestion(
path=str(suggestion.get("path") or "").strip(),
explanation=_clip(" ".join(str(suggestion.get("explanation") or "").split()), _MAX_EXPLANATION),
code=_clip(str(suggestion.get("code") or "").strip(), _MAX_CODE),
)
)
return built
def as_detail(suggestions: Any) -> list[dict[str, str]]:
"""Summarize suggestions for the audit event detail."""
return [
{"path": item.path, "explanation": item.explanation, "code": item.code}
for item in (suggestions or [])
]
def issue_section(suggestions: Any) -> str:
"""Render the suggestions as a markdown section for the incident issue.
Inputs: the CodeSuggestion list. Outputs: the section, or "" when there
are none.
The heading says "suggested", and the note says nothing was applied,
because a code block in an issue reads like a change that was made unless
it is told otherwise.
"""
items = list(suggestions or [])
if not items:
return ""
lines = [
"## Suggested fix (not applied)",
"Hermes could not open a pull request for this, so the change below was "
"not written, validated, or pushed anywhere. It is a starting point for "
"whoever picks this up, not a reviewed patch.",
]
for item in items:
lines.append("")
lines.append(f"**`{item.path}`** — {item.explanation}")
lines.append(f"{_FENCE}\n{item.code.replace(_FENCE, _FENCE_ESCAPE)}\n{_FENCE}")
return "\n".join(lines)
def _clip(value: str, limit: int) -> str:
"""Clip one field so a long suggestion cannot crowd out the diagnosis."""
return value if len(value) <= limit else value[:limit].rstrip() + _TRUNCATED

View File

@ -15,24 +15,12 @@ from __future__ import annotations
import re import re
from typing import Any from typing import Any
from ariadne.services import hermes_code_suggestion as code_suggestion_field
from ariadne.services import hermes_incident_evidence_section as evidence_section from ariadne.services import hermes_incident_evidence_section as evidence_section
from ariadne.services import hermes_suggested_remediation as suggestion_field from ariadne.services import hermes_suggested_remediation as suggestion_field
DEFAULT_MAX_BODY_CHARS = 8000 DEFAULT_MAX_BODY_CHARS = 8000
UNDIAGNOSED = "undiagnosed" UNDIAGNOSED = "undiagnosed"
# A finding is not a build failure, and rendering one through the other's
# wording produces an issue that is wrong in every heading: a line number
# printed as a build, a "first failed gate" for a build that passed, and a
# SonarQube link labelled "failed build".
SONAR_PREFIX = "sonar/"
def is_finding(context: dict) -> bool:
"""Report whether this issue describes a finding rather than a failure."""
return str(context.get("incident_id") or "").startswith(SONAR_PREFIX)
_MAX_TITLE_CHARS = 120 _MAX_TITLE_CHARS = 120
_MAX_FACTS = 10 _MAX_FACTS = 10
@ -49,7 +37,7 @@ _AUDIT_NOTE = (
"event types `hermes_autotriage_incident` and `hermes_autotriage_diagnosis`." "event types `hermes_autotriage_incident` and `hermes_autotriage_diagnosis`."
) )
_FOOTER_TEMPLATE = ( _FOOTER_TEMPLATE = (
"Filed automatically by Ariadne from a Hermes Agent diagnosis ({run}). " "Filed automatically by Ariadne from a Hermes Agent diagnosis (run `{run_id}`). "
"Hermes has no write access to this repository; no files or infrastructure were changed." "Hermes has no write access to this repository; no files or infrastructure were changed."
) )
# Some escalations never reach a model at all - a build still running has no # Some escalations never reach a model at all - a build still running has no
@ -105,10 +93,7 @@ def issue_title(context: dict) -> str:
`[hermes] {job} #{build}: {classification}` clipped to 120 characters. `[hermes] {job} #{build}: {classification}` clipped to 120 characters.
""" """
if is_finding(context): prefix = f"[hermes] {context.get('job')} #{context.get('build_number')}: "
prefix = f"[hermes] {context.get('job')}: "
else:
prefix = f"[hermes] {context.get('job')} #{context.get('build_number')}: "
classification = str(context.get("classification") or UNDIAGNOSED) classification = str(context.get("classification") or UNDIAGNOSED)
room = _MAX_TITLE_CHARS - len(prefix) room = _MAX_TITLE_CHARS - len(prefix)
if room <= len(_ELLIPSIS): if room <= len(_ELLIPSIS):
@ -137,7 +122,6 @@ def issue_body(context: dict, max_chars: int = DEFAULT_MAX_BODY_CHARS) -> str:
_facts_section(context), _facts_section(context),
evidence_section.evidence_section(context.get("bundle")), evidence_section.evidence_section(context.get("bundle")),
_inferences_section(context), _inferences_section(context),
code_suggestion_field.issue_section(context.get("code_suggestions")),
suggestion_field.issue_section(context.get("suggested_remediation")), suggestion_field.issue_section(context.get("suggested_remediation")),
_links_section(context), _links_section(context),
_footer(context), _footer(context),
@ -163,11 +147,6 @@ def _summary_line(context: dict) -> str:
incident_id = context.get("incident_id") incident_id = context.get("incident_id")
classification = context.get("classification") classification = context.get("classification")
if is_finding(context):
return (
f"SonarQube reports **{classification}** in `{context.get('finding_path')}`. "
"The build is green; this is a standing finding, not a failure."
)
run_id = str(context.get("run_id") or "") run_id = str(context.get("run_id") or "")
if not run_id or run_id == "unknown": if not run_id or run_id == "unknown":
# The headline is the first thing read, so it must not credit a model # The headline is the first thing read, so it must not credit a model
@ -187,27 +166,20 @@ def _footer(context: dict) -> str:
run_id = str(context.get("run_id") or "") run_id = str(context.get("run_id") or "")
if not run_id or run_id == "unknown": if not run_id or run_id == "unknown":
return _NO_RUN_FOOTER return _NO_RUN_FOOTER
# A bare id is something to copy; a link is a page to open, and that page return _FOOTER_TEMPLATE.format(run_id=run_id)
# is the whole answer to who decided this.
run_url = str(context.get("run_url") or "")
run = f"run [{run_id}]({run_url})" if run_url else f"run `{run_id}`"
return _FOOTER_TEMPLATE.format(run=run)
def _human_section(context: dict) -> str: def _human_section(context: dict) -> str:
"""Render the section explaining why the incident needs a person.""" """Render the section explaining why the incident needs a person."""
heading = "## What is wrong" if is_finding(context) else "## Why a human is needed" lines = ["## Why a human is needed", str(context.get("reason") or "no reason recorded")]
lines = [heading, str(context.get("reason") or "no reason recorded")]
observation = str(context.get("observation") or "").strip() observation = str(context.get("observation") or "").strip()
if observation: if observation:
# Escalations that never reached a model carry their explanation here # Escalations that never reached a model carry their explanation here
# rather than in facts, which are only populated from a diagnosis. # rather than in facts, which are only populated from a diagnosis.
lines.append(observation) lines.append(observation)
authorize_reason = str(context.get("authorize_reason") or "") authorize_reason = str(context.get("authorize_reason") or "")
if authorize_reason and is_finding(context): if authorize_reason:
lines.append(f"Ariadne opened no pull request for it: {authorize_reason}.")
elif authorize_reason:
lines.append(f"Ariadne did not authorize automated remediation: `{authorize_reason}`.") lines.append(f"Ariadne did not authorize automated remediation: `{authorize_reason}`.")
return "\n\n".join(lines) return "\n\n".join(lines)
@ -242,13 +214,10 @@ def _links_section(context: dict) -> str:
build_url = str(context.get("build_url") or "") build_url = str(context.get("build_url") or "")
proposal_url = str(context.get("code_proposal_url") or "") proposal_url = str(context.get("code_proposal_url") or "")
if is_finding(context): lines = [
lines = ["## Links", f"- SonarQube finding: {build_url}" if build_url else "- SonarQube finding: url unavailable"] "## Links",
else: f"- Failed build: {build_url}" if build_url else "- Failed build: url unavailable",
lines = [ ]
"## Links",
f"- Failed build: {build_url}" if build_url else "- Failed build: url unavailable",
]
if proposal_url: if proposal_url:
lines.append(f"- Proposed fix awaiting review: {proposal_url}") lines.append(f"- Proposed fix awaiting review: {proposal_url}")
lines.append(f"- {_AUDIT_NOTE}") lines.append(f"- {_AUDIT_NOTE}")

View File

@ -22,7 +22,7 @@ from typing import Any
import httpx import httpx
from ..utils.logging import get_logger from ..utils.logging import get_logger
from . import hermes_code_repair, hermes_incident_body as body from . import hermes_incident_body as body
logger = get_logger(__name__) logger = get_logger(__name__)
@ -171,9 +171,7 @@ def issue_config(config: Any) -> dict[str, Any]:
} }
def issue_context( def issue_context(base: dict[str, Any], diagnosis: dict[str, Any]) -> dict[str, Any]:
base: dict[str, Any], diagnosis: dict[str, Any], hermes_ui_url: str = ""
) -> dict[str, Any]:
"""Flatten the incident, evidence bundle, and outcome into body inputs. """Flatten the incident, evidence bundle, and outcome into body inputs.
Inputs: the incident identity fields and the diagnosis dict passed to Inputs: the incident identity fields and the diagnosis dict passed to
@ -205,7 +203,6 @@ def issue_context(
"facts": [body.fact_fields(fact) for fact in getattr(decision, "facts", None) or []], "facts": [body.fact_fields(fact) for fact in getattr(decision, "facts", None) or []],
"inferences": list(getattr(decision, "inferences", None) or []), "inferences": list(getattr(decision, "inferences", None) or []),
"suggested_remediation": getattr(decision, "suggested_remediation", None), "suggested_remediation": getattr(decision, "suggested_remediation", None),
"code_suggestions": getattr(decision, "code_suggestions", None) or [],
# The raw excerpt, so the issue shows the failure instead of citing # The raw excerpt, so the issue shows the failure instead of citing
# where in Jenkins the failure can be found. # where in Jenkins the failure can be found.
"bundle": bundle, "bundle": bundle,
@ -214,7 +211,6 @@ def issue_context(
# console text is the whole explanation and must not be dropped. # console text is the whole explanation and must not be dropped.
"observation": _observation(jenkins) if decision is None else "", "observation": _observation(jenkins) if decision is None else "",
"run_id": diagnosis.get("run_id"), "run_id": diagnosis.get("run_id"),
"run_url": hermes_code_repair.run_url(hermes_ui_url, str(diagnosis.get("run_id") or "")),
"code_proposal_url": str(proposal.get("url") or "") if isinstance(proposal, dict) else "", "code_proposal_url": str(proposal.get("url") or "") if isinstance(proposal, dict) else "",
} }
@ -244,7 +240,7 @@ def _file_issue(
if not cfg["enabled"] or target is None or _filed_count(tick_state) >= _max_per_tick(cfg): if not cfg["enabled"] or target is None or _filed_count(tick_state) >= _max_per_tick(cfg):
return None return None
repo_cfg = {**cfg, "owner": target[0], "repo": target[1]} repo_cfg = {**cfg, "owner": target[0], "repo": target[1]}
context = issue_context(base, diagnosis, str(getattr(config, "hermes_ui_url", "") or "")) context = issue_context(base, diagnosis)
existing = find_open_incident_issue( existing = find_open_incident_issue(
repo_cfg, job, str(context["classification"]), str(context["incident_id"]) repo_cfg, job, str(context["classification"]), str(context["incident_id"])
) )

View File

@ -1,250 +0,0 @@
"""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": "<must equal __INCIDENT_ID__>", "analysis": "<what is actually wrong, in one or two sentences>", "code_suggestions": [{"path": "<repository-relative file>", "explanation": "<what is wrong there and why this change fixes it>", "code": "<the suggested code>"}], "human_required": <bool>, "reason": "<string>"}
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": "",
"finding_path": issue.get("path"),
"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

View File

@ -149,38 +149,6 @@ def component_path(component: Any) -> str:
return path.strip() if separator else "" return path.strip() if separator else ""
INCIDENT_PREFIX = "sonar/"
# The console opens one finding in place, with the rule, the effort estimate
# and the offending lines highlighted.
_ISSUE_PATH = "/project/issues?resolved=false&id={project}&open={key}"
def issue_url(incident_id: str, ui_url: str) -> str:
"""Build the console link for the finding an incident came from.
Inputs: an incident id shaped `sonar/<project>/<key>`, and the SonarQube
base url. Outputs: the deep link, or "" for any incident that did not come
from a sweep - a build-driven repair has no finding to point at.
Parsed from the incident id rather than threaded through the flow: the id
already carries both halves, and inventing a second path for them to travel
is a second thing that can disagree with the first.
"""
base = str(ui_url or "").rstrip("/")
text = str(incident_id or "")
if not base or not text.startswith(INCIDENT_PREFIX):
return ""
# `sonar/<project>/<rule>/<key>`; the rule sits between them so the
# project is read from the front and the key from the back.
remainder = text[len(INCIDENT_PREFIX) :]
project, separator, rest = remainder.partition("/")
_, _, key = rest.rpartition("/") if "/" in rest else ("", "", rest)
if not separator or not project.strip() or not key.strip():
return ""
return base + _ISSUE_PATH.format(project=project.strip(), key=key.strip())
def _types(cfg: dict) -> tuple[str, ...]: def _types(cfg: dict) -> tuple[str, ...]:
"""Return the finding types this deployment allows, defaulting to all.""" """Return the finding types this deployment allows, defaulting to all."""

View File

@ -27,7 +27,7 @@ from typing import Any
from ..settings import settings from ..settings import settings
from ..utils.logging import get_logger from ..utils.logging import get_logger
from . import hermes_code_flow, hermes_code_repair, hermes_sonar_advice, hermes_sonar_client from . import hermes_code_flow, hermes_sonar_client
logger = get_logger(__name__) logger = get_logger(__name__)
@ -84,7 +84,7 @@ def sweep(storage: Any, config: Any, hermes_cfg: dict) -> dict[str, Any]:
"hermes sonar sweep failed", "hermes sonar sweep failed",
extra={"event": SWEEP_EVENT_TYPE, "status": "error", "detail": str(exc)}, extra={"event": SWEEP_EVENT_TYPE, "status": "error", "detail": str(exc)},
) )
return {"proposed": 0, "advised": 0, "skipped": [f"sweep_failed: {exc}"], "projects": 0} return {"proposed": 0, "skipped": [f"sweep_failed: {exc}"], "projects": 0}
def _sweep(storage: Any, config: Any, hermes_cfg: dict) -> dict[str, Any]: def _sweep(storage: Any, config: Any, hermes_cfg: dict) -> dict[str, Any]:
@ -92,12 +92,11 @@ def _sweep(storage: Any, config: Any, hermes_cfg: dict) -> dict[str, Any]:
projects = _projects(config) projects = _projects(config)
if not projects: if not projects:
return {"proposed": 0, "advised": 0, "skipped": [_NO_PROJECTS], "projects": 0} return {"proposed": 0, "skipped": [_NO_PROJECTS], "projects": 0}
sonar_cfg = client_config(config) sonar_cfg = client_config(config)
code_cfg = hermes_code_flow.code_config(config) code_cfg = hermes_code_flow.code_config(config)
budget = _max_per_sweep(config) budget = _max_per_sweep(config)
proposed = 0 proposed = 0
advised = 0
skipped: list[str] = [] skipped: list[str] = []
for project, job in sorted(projects.items()): for project, job in sorted(projects.items()):
if proposed >= budget: if proposed >= budget:
@ -107,13 +106,12 @@ def _sweep(storage: Any, config: Any, hermes_cfg: dict) -> dict[str, Any]:
storage, project, job, sonar_cfg, code_cfg, hermes_cfg, config storage, project, job, sonar_cfg, code_cfg, hermes_cfg, config
) )
proposed += outcome["proposed"] proposed += outcome["proposed"]
advised += outcome.get("advised", 0)
skipped.extend(outcome["skipped"]) skipped.extend(outcome["skipped"])
logger.info( logger.info(
"hermes sonar sweep finished", "hermes sonar sweep finished",
extra={"event": SWEEP_EVENT_TYPE, "status": "ok", "detail": f"proposed={proposed}"}, extra={"event": SWEEP_EVENT_TYPE, "status": "ok", "detail": f"proposed={proposed}"},
) )
return {"proposed": proposed, "advised": advised, "skipped": skipped, "projects": len(projects)} return {"proposed": proposed, "skipped": skipped, "projects": len(projects)}
def _propose_for_project( # noqa: PLR0913 - one project needs every config the flow does def _propose_for_project( # noqa: PLR0913 - one project needs every config the flow does
@ -133,114 +131,28 @@ def _propose_for_project( # noqa: PLR0913 - one project needs every config the
issues, error = hermes_sonar_client.fetch_issues(sonar_cfg, project) issues, error = hermes_sonar_client.fetch_issues(sonar_cfg, project)
if error: if error:
return {"proposed": 0, "skipped": [f"{project}: {error}"]} return {"proposed": 0, "skipped": [f"{project}: {error}"]}
open_rules = proposed_rules(repo_cfg, project) chosen = select_issue(issues, repo_cfg, config)
chosen = select_issue(issues, repo_cfg, config, open_rules) if chosen is None:
if chosen is not None: return {"proposed": 0, "skipped": [f"{project}: no mechanically fixable finding"]}
result = hermes_code_flow.propose_code_fix( result = hermes_code_flow.propose_code_fix(
storage, storage,
incident_id=incident_id(project, chosen), incident_id=incident_id(project, chosen),
job=job, job=job,
build_number=f"sonar-{chosen['key']}", build_number=f"sonar-{chosen['key']}",
bundle=bundle_for(project, chosen), bundle=bundle_for(project, chosen),
hermes_cfg=hermes_cfg, hermes_cfg=hermes_cfg,
code_cfg=code_cfg, code_cfg=code_cfg,
) )
if result.get("status") == "pr_opened": if result.get("status") == "pr_opened":
return {"proposed": 1, "skipped": []} return {"proposed": 1, "skipped": []}
# A declined pull request is not a dead end. The finding is still real return {"proposed": 0, "skipped": [f"{project}: {result.get('reason')}"]}
# 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 def select_issue(issues: list[dict], repo_cfg: dict, config: Any) -> dict[str, Any] | None:
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]:
"""Return the rules this project already has an open proposal for.
Inputs: the resolved per-repo cfg and the SonarQube project key. Outputs:
the rule ids, read back from the open pull requests' incident ids.
One rule is usually one root cause spread across many files: S2208 appears
in three Ariadne modules, and the cognitive-complexity rule in dozens. One
finding at a time with no memory would open a near-identical pull request
for every instance, and a reviewer facing thirty of those reads none of
them. Until the open one is dealt with, the rest of that rule waits.
Fails open, like every other duplicate check here: an unreadable list
yields an empty set, so the cost of a lookup failure is one extra proposal
rather than a whole rule silently going unreported.
"""
incidents, _error = hermes_code_repair.open_proposal_incidents(repo_cfg)
prefix = f"{hermes_sonar_client.INCIDENT_PREFIX}{project}/"
rules = set()
for incident in incidents:
if not incident.startswith(prefix):
continue
rule, separator, _key = incident[len(prefix) :].rpartition("/")
if separator and rule:
rules.add(rule)
return rules
def select_issue(
issues: list[dict], repo_cfg: dict, config: Any, open_rules: set[str] | None = None
) -> dict[str, Any] | None:
"""Pick the single most mechanically fixable finding for one project. """Pick the single most mechanically fixable finding for one project.
Inputs: the normalized findings, the resolved per-repo cfg (whose write Inputs: the normalized findings, the resolved per-repo cfg (whose write
allowlist decides what is patchable at all), the settings object, and the allowlist decides what is patchable at all), and the settings object.
rules that already have an open proposal.
Outputs: one finding, or None when none qualify. Outputs: one finding, or None when none qualify.
Ordered by effort, then severity, then key. Effort leads because it is the Ordered by effort, then severity, then key. Effort leads because it is the
@ -251,12 +163,10 @@ def select_issue(
""" """
ceiling = _max_effort_minutes(config) ceiling = _max_effort_minutes(config)
seen = open_rules or set()
eligible = [ eligible = [
issue issue
for issue in issues for issue in issues
if str(issue.get("rule") or "") not in seen if _is_writable(str(issue.get("path") or ""), repo_cfg)
and _is_writable(str(issue.get("path") or ""), repo_cfg)
and effort_minutes(issue.get("effort")) is not None and effort_minutes(issue.get("effort")) is not None
and (effort_minutes(issue.get("effort")) or 0) <= ceiling and (effort_minutes(issue.get("effort")) or 0) <= ceiling
] ]
@ -326,14 +236,9 @@ def bundle_for(project: str, issue: dict[str, Any]) -> dict[str, Any]:
def incident_id(project: str, issue: dict[str, Any]) -> str: def incident_id(project: str, issue: dict[str, Any]) -> str:
"""Name the incident for one finding, stable across sweeps. """Name the incident for one finding, stable across sweeps."""
Carries the rule as well as the key so an open proposal announces which return f"sonar/{project}/{issue.get('key')}"
root cause is already being reviewed. The alternative was a second index
of proposed rules, which could disagree with the pull requests themselves.
"""
return f"sonar/{project}/{issue.get('rule')}/{issue.get('key')}"
def client_config(config: Any) -> dict[str, Any]: def client_config(config: Any) -> dict[str, Any]:

View File

@ -26,7 +26,7 @@ jenkins.first_failed_stage names the pipeline stage that failed when the build r
The jenkins.console_failures array holds excerpts around detected failure markers in chronological order; the earliest region usually contains the first enforced failure, and jenkins.console_tail is the end of the build which often only shows downstream noise. The jenkins.console_failures array holds excerpts around detected failure markers in chronological order; the earliest region usually contains the first enforced failure, and jenkins.console_tail is the end of the build which often only shows downstream noise.
Distinguish facts from inference. Distinguish facts from inference.
Return ONLY a single JSON object with exactly these keys and no others: Return ONLY a single JSON object with exactly these keys and no others:
{"incident_id": "<must equal __INCIDENT_ID__>", "classification": "<short snake_case name for what actually failed>", "confidence": <0..1>, "facts": [{"statement": "...", "source": "jenkins|opensearch|victoriametrics|kubernetes|flux|gitea", "reference": "..."}], "inferences": ["..."], "first_failed_gate": "<string>", "requested_action": <object or null>, "suggested_remediation": <object or null>, "code_suggestions": <array or null>, "human_required": <bool>, "reason": "<string>"} {"incident_id": "<must equal __INCIDENT_ID__>", "classification": "<short snake_case name for what actually failed>", "confidence": <0..1>, "facts": [{"statement": "...", "source": "jenkins|opensearch|victoriametrics|kubernetes|flux|gitea", "reference": "..."}], "inferences": ["..."], "first_failed_gate": "<string>", "requested_action": <object or null>, "suggested_remediation": <object or null>, "human_required": <bool>, "reason": "<string>"}
You are diagnosing only; you do not execute anything. Ariadne separately validates and executes any requested action under its own authorization policy, and will refuse anything its own reading of the evidence does not support. You are diagnosing only; you do not execute anything. Ariadne separately validates and executes any requested action under its own authorization policy, and will refuse anything its own reading of the evidence does not support.
Write the reason and the inferences for an engineer who maintains this service and who has no knowledge of how this triage system is configured. Describe the failure and what a fix would involve. Do not discuss classifications, actions, policies, or which of them are permitted; those are Ariadne's concern and are meaningless in the service's issue tracker. Write the reason and the inferences for an engineer who maintains this service and who has no knowledge of how this triage system is configured. Describe the failure and what a fix would involve. Do not discuss classifications, actions, policies, or which of them are permitted; those are Ariadne's concern and are meaningless in the service's issue tracker.
Three classifications have a predefined remediation. Use one only when the evidence plainly shows that failure; otherwise leave requested_action null. Three classifications have a predefined remediation. Use one only when the evidence plainly shows that failure; otherwise leave requested_action null.
@ -35,7 +35,6 @@ Use workspace_storage_exhausted with requested_action {"type": "run_ariadne_job"
Use jenkins_agent_provisioning_failure with requested_action {"type": "run_ariadne_job", "id": "clear_stuck_agent_pods"} when the evidence shows the build never got an agent (all nodes of a label offline, an agent pod stuck ContainerCreating or Pending, or an error in provisioning) rather than failing once it was running. Use jenkins_agent_provisioning_failure with requested_action {"type": "run_ariadne_job", "id": "clear_stuck_agent_pods"} when the evidence shows the build never got an agent (all nodes of a label offline, an agent pod stuck ContainerCreating or Pending, or an error in provisioning) rather than failing once it was running.
Otherwise leave requested_action null. Otherwise leave requested_action null.
When you leave requested_action null because none of the remediations above fits, and you can name a remediation that a maintainer could reasonably automate for this failure, set suggested_remediation to {"action_id": "<snake_case name for the remediation you are proposing>", "summary": "<what it would do, in one or two sentences>", "evidence_required": "<what would have to be present in the evidence before running it is safe>"}; otherwise set it to null. This proposes work for a maintainer to build and does not request anything: nothing you name here can be executed, and it must be null whenever requested_action is set. Propose one only when the same remediation would be correct for any build failing this way, not merely for this build. When you leave requested_action null because none of the remediations above fits, and you can name a remediation that a maintainer could reasonably automate for this failure, set suggested_remediation to {"action_id": "<snake_case name for the remediation you are proposing>", "summary": "<what it would do, in one or two sentences>", "evidence_required": "<what would have to be present in the evidence before running it is safe>"}; otherwise set it to null. This proposes work for a maintainer to build and does not request anything: nothing you name here can be executed, and it must be null whenever requested_action is set. Propose one only when the same remediation would be correct for any build failing this way, not merely for this build.
When the failure is in this repository's own source or tests and you can see what the change should be, set code_suggestions to at most three {"path": "<repository-relative file>", "explanation": "<what is wrong there and why this change fixes it>", "code": "<the suggested code>"}; otherwise set it to null. Write them for a maintainer who will read the change and then make it themselves: nothing you put here is applied, validated against the file, or committed anywhere. Suggest the fix you actually believe is right even when it spans several places or is too large to express as one edit, and prefer correcting the code under test over changing a test's expectations.
Set human_required to true when decisive evidence is missing or the failure needs a judgement only a maintainer can make; otherwise set it false and say plainly what you believe is wrong. Set human_required to true when decisive evidence is missing or the failure needs a judgement only a maintainer can make; otherwise set it false and say plainly what you believe is wrong.
Do not perform mutations.""" Do not perform mutations."""

View File

@ -319,13 +319,11 @@ class Settings:
hermes_sonar_cron: str hermes_sonar_cron: str
hermes_sonar_enabled: bool hermes_sonar_enabled: bool
hermes_sonar_url: str hermes_sonar_url: str
hermes_sonar_ui_url: str
hermes_sonar_token: str hermes_sonar_token: str
hermes_sonar_projects: dict[str, str] hermes_sonar_projects: dict[str, str]
hermes_sonar_types: list[str] hermes_sonar_types: list[str]
hermes_sonar_severities: list[str] hermes_sonar_severities: list[str]
hermes_sonar_max_per_sweep: int hermes_sonar_max_per_sweep: int
hermes_sonar_advice_enabled: bool
hermes_sonar_max_effort_minutes: int hermes_sonar_max_effort_minutes: int
hermes_sonar_timeout_seconds: float hermes_sonar_timeout_seconds: float

View File

@ -60,7 +60,6 @@ def _hermes_autotriage_config() -> dict[str, Any]:
"ARIADNE_HERMES_SONAR_URL", "http://sonarqube.quality.svc.cluster.local:9000" "ARIADNE_HERMES_SONAR_URL", "http://sonarqube.quality.svc.cluster.local:9000"
).rstrip("/"), ).rstrip("/"),
"hermes_sonar_token": _env("ARIADNE_HERMES_SONAR_TOKEN", ""), "hermes_sonar_token": _env("ARIADNE_HERMES_SONAR_TOKEN", ""),
"hermes_sonar_ui_url": _env("ARIADNE_HERMES_SONAR_UI_URL", ""),
"hermes_sonar_projects": _pair_map(_env("ARIADNE_HERMES_SONAR_PROJECTS", "")), "hermes_sonar_projects": _pair_map(_env("ARIADNE_HERMES_SONAR_PROJECTS", "")),
"hermes_sonar_types": [ "hermes_sonar_types": [
item.strip() item.strip()
@ -73,7 +72,6 @@ def _hermes_autotriage_config() -> dict[str, Any]:
if item.strip() if item.strip()
], ],
"hermes_sonar_max_per_sweep": _env_int("ARIADNE_HERMES_SONAR_MAX_PER_SWEEP", 1), "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_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_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_namespace": _env("ARIADNE_HERMES_DEMO_NAMESPACE", "hermes-triage-demo"),

View File

@ -37,7 +37,6 @@ def _hermes_cfg() -> dict:
def _code_cfg(**overrides) -> dict: # type: ignore[no-untyped-def] def _code_cfg(**overrides) -> dict: # type: ignore[no-untyped-def]
base = { base = {
"hermes_ui_url": "", "hermes_ui_url": "",
"sonar_ui_url": "",
"candidate_path": "src/discount.py", "candidate_path": "src/discount.py",
"allowed_path_prefixes": ["src/"], "allowed_path_prefixes": ["src/"],
"allowed_suffixes": [".py"], "allowed_suffixes": [".py"],

View File

@ -3,8 +3,6 @@ from __future__ import annotations
import base64 import base64
import json import json
import httpx
from ariadne.services import hermes_code_repair as module from ariadne.services import hermes_code_repair as module
from ariadne.services.hermes_code_patch import ProposedPatch from ariadne.services.hermes_code_patch import ProposedPatch
@ -365,11 +363,9 @@ def test_the_pull_request_names_the_hermes_run(monkeypatch) -> None:
) )
body = calls["requests"][0][2]["json"]["body"] body = calls["requests"][0][2]["json"]["body"]
assert ( assert "**Hermes run:** `run_a5af87af`" in body
"**Hermes run:** [run_a5af87af]" assert "https://agent.bstein.dev." in body
"(https://agent.bstein.dev/chat?resume=run_a5af87af)" assert "every tool call it made" in body
) in body
assert "the tools it called" in body
def test_the_pull_request_omits_the_link_when_no_ui_is_configured(monkeypatch) -> None: def test_the_pull_request_omits_the_link_when_no_ui_is_configured(monkeypatch) -> None:
@ -390,90 +386,3 @@ def test_the_pull_request_stays_readable_without_a_run_id(monkeypatch) -> None:
body = calls["requests"][0][2]["json"]["body"] body = calls["requests"][0][2]["json"]["body"]
assert "Hermes run" not in body assert "Hermes run" not in body
assert "requires human review" in body assert "requires human review" in body
def test_the_run_link_reopens_the_run_in_the_hermes_console() -> None:
"""One helper, so the pull request and the demo monitor cannot diverge."""
assert module.run_url("https://agent.bstein.dev/", "run_x") == (
"https://agent.bstein.dev/chat?resume=run_x"
)
assert module.run_url("", "run_x") == ""
assert module.run_url("https://agent.bstein.dev", " ") == ""
def test_a_sweep_proposal_links_to_the_finding_that_caused_it(monkeypatch) -> None:
"""The reviewer's first question is what the finding said."""
calls = _install_http(monkeypatch, [FakeResponse(201, {"number": 6, "html_url": "u"})])
module.open_pull_request(
{**_cfg(), "sonar_ui_url": "https://quality.bstein.dev"},
"sonar/ariadne/AZ2y0FYFKy9i4pkIpNlV",
"run_x",
BRANCH,
_patch(),
"analysis",
)
body = calls["requests"][0][2]["json"]["body"]
assert "**SonarQube finding:** https://quality.bstein.dev/project/issues" in body
assert "open=AZ2y0FYFKy9i4pkIpNlV" in body
def test_a_build_driven_proposal_has_no_finding_line(monkeypatch) -> None:
calls = _install_http(monkeypatch, [FakeResponse(201, {"number": 6, "html_url": "u"})])
module.open_pull_request(
{**_cfg(), "sonar_ui_url": "https://quality.bstein.dev"},
INCIDENT_ID,
"run_x",
BRANCH,
_patch(),
"analysis",
)
assert "SonarQube finding" not in calls["requests"][0][2]["json"]["body"]
def test_open_proposal_incidents_reads_them_from_the_titles(monkeypatch) -> None:
"""The pull requests are the thing that exists; an index could disagree."""
calls = _install_http(
monkeypatch,
[
FakeResponse(
200,
[
{"title": "fix(hermes): repair sonar/ariadne/python:S2208/AZ1"},
{"title": "fix(hermes): repair ariadne/408"},
{"title": "chore: something a person opened"},
{"title": ""},
"not-a-dict",
],
)
],
)
incidents, error = module.open_proposal_incidents(_cfg())
assert error is None
assert incidents == ["sonar/ariadne/python:S2208/AZ1", "ariadne/408"]
assert calls["requests"][0][2]["params"]["state"] == "open"
def test_open_proposal_incidents_fails_open(monkeypatch) -> None:
"""A duplicate is noise; a suppressed rule is lost work."""
_install_http(monkeypatch, [FakeResponse(500, None)])
assert module.open_proposal_incidents(_cfg()) == ([], "open proposal lookup http 500")
_install_http(monkeypatch, [FakeResponse(200, {"not": "a list"})])
assert module.open_proposal_incidents(_cfg()) == ([], "open proposal payload is not a list")
_install_http(monkeypatch, [httpx.ConnectError("refused")])
incidents, error = module.open_proposal_incidents(_cfg())
assert incidents == []
assert "open proposal lookup failed" in error
assert module.open_proposal_incidents({**_cfg(), "gitea_base_url": ""}) == (
[],
"gitea base url is empty",
)

View File

@ -1,174 +0,0 @@
"""Tests for code suggestions carried by an escalated diagnosis."""
from __future__ import annotations
import json
import pytest
from ariadne.services import hermes_autotriage_decision as decision_module
from ariadne.services import hermes_code_suggestion as module
from ariadne.services import hermes_incident_body as body
SUGGESTION = {
"path": "ariadne/utils/errors.py",
"explanation": "safe_error_detail drops the response body, so the assertion never sees it.",
"code": 'def safe_error_detail(exc):\n return f"http {exc.response.status_code}: {exc.response.text}"',
}
def _payload(**overrides):
payload = {
"incident_id": "ariadne/408",
"classification": "pytest_test_failure",
"confidence": 0.9,
"facts": [],
"inferences": [],
"first_failed_gate": "tests",
"requested_action": None,
"human_required": True,
"reason": "a repository test failure",
}
payload.update(overrides)
return payload
def _parse(payload):
return decision_module.parse_triage_response(json.dumps(payload), "ariadne/408")
def test_a_response_without_the_field_still_validates() -> None:
"""Responses written before the field existed must keep parsing."""
outcome = _parse(_payload())
assert outcome.valid
assert outcome.decision.code_suggestions == []
def test_suggestions_are_parsed() -> None:
outcome = _parse(_payload(code_suggestions=[dict(SUGGESTION)]))
suggestion = outcome.decision.code_suggestions[0]
assert suggestion.path == "ariadne/utils/errors.py"
assert "drops the response body" in suggestion.explanation
assert "def safe_error_detail" in suggestion.code
def test_a_null_field_is_accepted() -> None:
assert _parse(_payload(code_suggestions=None)).decision.code_suggestions == []
@pytest.mark.parametrize(
("suggestions", "expected"),
[
("nope", "must be a list or null"),
([dict(SUGGESTION)] * 4, "at most 3 suggestions"),
(["not-an-object"], "[0] must be an object"),
([{"path": "a"}], "[0] must have exactly path, explanation, code"),
([{**SUGGESTION, "code": " "}], "[0] code must be a non-empty string"),
([{**SUGGESTION, "path": 7}], "[0] path must be a non-empty string"),
],
)
def test_a_malformed_field_is_rejected(suggestions, expected) -> None:
outcome = _parse(_payload(code_suggestions=suggestions))
assert not outcome.valid
assert expected in outcome.reject_reason
def test_a_long_suggestion_is_clipped() -> None:
outcome = _parse(
_payload(code_suggestions=[{**SUGGESTION, "code": "x" * 5000, "explanation": "y " * 900}])
)
suggestion = outcome.decision.code_suggestions[0]
assert len(suggestion.code) <= 1700
assert len(suggestion.explanation) <= 700
def test_a_suggestion_never_becomes_an_action() -> None:
"""It is advice in an issue; the gates must not see it as anything else."""
outcome = _parse(_payload(code_suggestions=[dict(SUGGESTION)], human_required=False))
allowed, reason = decision_module.authorize_action(
outcome,
{
"allowed_actions": ["retry_transient_infra"],
"action_classifications": {"pytest_test_failure": "retry_transient_infra"},
"autoremediation_enabled": True,
"min_confidence": 0.5,
},
prior_action_count=0,
build_is_terminal_failure=True,
job_allowlisted=True,
evidence_has_signature=True,
)
assert not allowed
assert reason == "requested_action_missing"
def test_the_issue_section_says_the_change_was_not_applied() -> None:
"""A code block in an issue reads as a change that was made."""
section = module.issue_section(module.from_payload({"code_suggestions": [dict(SUGGESTION)]}))
assert "## Suggested fix (not applied)" in section
assert "not written, validated, or pushed anywhere" in section
assert "**`ariadne/utils/errors.py`**" in section
assert section.count("```") == 2
assert module.issue_section(None) == ""
assert module.issue_section([]) == ""
def test_a_fence_inside_a_suggestion_cannot_break_the_block() -> None:
section = module.issue_section(
module.from_payload({"code_suggestions": [{**SUGGESTION, "code": "a\n```\nb"}]})
)
assert section.count("```") == 2
def test_the_detail_recorded_for_the_audit_trail() -> None:
detail = module.as_detail(module.from_payload({"code_suggestions": [dict(SUGGESTION)]}))
assert detail[0]["path"] == "ariadne/utils/errors.py"
assert module.as_detail(None) == []
def test_from_payload_tolerates_junk() -> None:
assert module.from_payload({}) == []
assert module.from_payload({"code_suggestions": "nope"}) == []
assert module.from_payload({"code_suggestions": ["x"]}) == []
def test_the_issue_body_carries_the_suggestion_and_keeps_its_marker() -> None:
rendered = body.issue_body(
{
"incident_id": "ariadne/408",
"job": "ariadne",
"build_number": 408,
"classification": "pytest_test_failure",
"reason": "a repository test failure",
"run_id": "run-1",
"code_suggestions": module.from_payload({"code_suggestions": [dict(SUGGESTION)]}),
}
)
assert "## Suggested fix (not applied)" in rendered
assert "def safe_error_detail" in rendered
assert rendered.index("Suggested fix") < rendered.index("## Links")
assert rendered.rstrip().endswith("-->")
def test_the_prompt_asks_for_suggestions_without_promising_to_apply_them() -> None:
from ariadne.services import hermes_triage_prompt
prompt = hermes_triage_prompt.build_prompt("ariadne/408", "ariadne", {})
assert '"code_suggestions": <array or null>' in prompt
assert "at most three" in prompt
assert "nothing you put here is applied" in prompt
assert "prefer correcting the code under test" in prompt

View File

@ -7,23 +7,6 @@ from types import SimpleNamespace
from ariadne.services import hermes_incident_body as body from ariadne.services import hermes_incident_body as body
from ariadne.services import hermes_incident_issue as module from ariadne.services import hermes_incident_issue as module
_BASE = {"incident_id": "ariadne/408", "job": "ariadne", "build_number": 408}
def _diagnosis(**overrides):
"""A completed diagnosis, as maybe_file_issue receives it."""
diagnosis = {
"bundle": {"jenkins": {"url": "https://ci.example/job/ariadne/408/"}},
"outcome": None,
"authorize_reason": "human_required",
"run_id": "run_a5af87af",
}
diagnosis.update(overrides)
return diagnosis
def test_body_does_not_claim_a_diagnosis_that_never_happened() -> None: def test_body_does_not_claim_a_diagnosis_that_never_happened() -> None:
"""A hung build is escalated without any model call. """A hung build is escalated without any model call.
@ -116,21 +99,3 @@ def test_explicit_classification_is_used_only_without_a_decision() -> None:
"classification": "build_exceeded_time_cap", "run_id": "run_x"}, "classification": "build_exceeded_time_cap", "run_id": "run_x"},
) )
assert ctx["classification"] == "pytest_test_failure" assert ctx["classification"] == "pytest_test_failure"
def test_the_issue_footer_links_to_the_hermes_run() -> None:
"""A bare id is something to copy; a link is a page to open."""
context = module.issue_context(_BASE, _diagnosis(), "https://agent.bstein.dev")
rendered = body.issue_body(context)
assert context["run_url"].startswith("https://agent.bstein.dev/chat?resume=")
assert f"run [{context['run_id']}]({context['run_url']})" in rendered
def test_the_footer_falls_back_to_a_bare_id_without_a_console() -> None:
context = module.issue_context(_BASE, _diagnosis())
rendered = body.issue_body(context)
assert context["run_url"] == ""
assert f"run `{context['run_id']}`" in rendered

View File

@ -493,4 +493,3 @@ def test_broken_repo_map_entries_file_nothing(monkeypatch) -> None:
is None is None
) )
assert calls["requests"] == [] assert calls["requests"] == []

View File

@ -1,289 +0,0 @@
"""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
def test_a_finding_issue_never_uses_build_failure_wording() -> None:
"""A line number printed as a build number reads as a failure that never happened."""
from ariadne.services import hermes_incident_body as body
from ariadne.services import hermes_code_suggestion as cs
context = {
"incident_id": INCIDENT,
"job": "ariadne",
"build_number": 42,
"classification": "python:S3776",
"finding_path": "ariadne/services/thing.py",
"reason": "The function branches five ways.",
"authorize_reason": "no automated patch was possible for this finding",
"build_url": "https://quality.example/project/issues?open=AZ1",
"run_id": "run_x",
"run_url": "https://agent.example/chat?resume=run_x",
"code_suggestions": cs.from_payload({"code_suggestions": [dict(SUGGESTION)]}),
}
assert body.issue_title(context) == "[hermes] ariadne: python:S3776"
rendered = body.issue_body(context)
assert rendered.startswith("SonarQube reports **python:S3776**")
assert "The build is green" in rendered
assert "## What is wrong" in rendered
assert "- SonarQube finding: https://quality.example" in rendered
assert "Failed build" not in rendered
assert "first failed gate" not in rendered
assert "Why a human is needed" not in rendered
def test_a_build_failure_issue_is_unchanged() -> None:
"""The finding wording must not leak into real triage issues."""
from ariadne.services import hermes_incident_body as body
context = {
"incident_id": "ariadne/408",
"job": "ariadne",
"build_number": 408,
"classification": "pytest_test_failure",
"reason": "a repository test failure",
"authorize_reason": "human_required",
"build_url": "https://ci.example/job/ariadne/408/",
"run_id": "run_y",
}
assert body.issue_title(context) == "[hermes] ariadne #408: pytest_test_failure"
rendered = body.issue_body(context)
assert "## Why a human is needed" in rendered
assert "- Failed build: https://ci.example" in rendered
assert "did not authorize automated remediation" in rendered
assert "SonarQube" not in rendered

View File

@ -248,27 +248,3 @@ def test_the_timeout_falls_back_when_unparseable() -> None:
) )
def test_the_component_key_yields_the_repository_path(component, expected) -> None: def test_the_component_key_yields_the_repository_path(component, expected) -> None:
assert module.component_path(component) == expected assert module.component_path(component) == expected
@pytest.mark.parametrize(
("incident", "expected"),
[
(
"sonar/ariadne/AZ2y0FYFKy9i4pkIpNlV",
"https://quality.bstein.dev/project/issues"
"?resolved=false&id=ariadne&open=AZ2y0FYFKy9i4pkIpNlV",
),
# A build-driven repair has no finding to point at.
("ariadne/408", ""),
("sonar/ariadne", ""),
("sonar//key", ""),
("sonar/proj/ ", ""),
("", ""),
],
)
def test_the_finding_link_is_derived_from_the_incident_id(incident, expected) -> None:
assert module.issue_url(incident, "https://quality.bstein.dev/") == expected
def test_no_finding_link_without_a_configured_console() -> None:
assert module.issue_url("sonar/ariadne/AZ1", "") == ""

View File

@ -42,7 +42,6 @@ def _config(**overrides):
"hermes_sonar_max_per_sweep": 1, "hermes_sonar_max_per_sweep": 1,
"hermes_sonar_max_effort_minutes": 20, "hermes_sonar_max_effort_minutes": 20,
"hermes_sonar_timeout_seconds": 5, "hermes_sonar_timeout_seconds": 5,
"hermes_sonar_advice_enabled": False,
} }
values.update(overrides) values.update(overrides)
return SimpleNamespace(**values) return SimpleNamespace(**values)
@ -73,10 +72,6 @@ def wiring(monkeypatch):
return calls["result"] return calls["result"]
monkeypatch.setattr(module.hermes_code_flow, "propose_code_fix", _propose) monkeypatch.setattr(module.hermes_code_flow, "propose_code_fix", _propose)
monkeypatch.setattr(
module.hermes_code_repair, "open_proposal_incidents",
lambda cfg: (calls.get("open_incidents", []), None),
)
return calls return calls
@ -149,9 +144,9 @@ def test_a_sweep_opens_one_proposal_for_the_chosen_finding(wiring) -> None:
result = module.sweep(storage, _config(), {}) result = module.sweep(storage, _config(), {})
assert result == {"proposed": 1, "advised": 0, "skipped": [], "projects": 1} assert result == {"proposed": 1, "skipped": [], "projects": 1}
proposal = wiring["proposals"][0] proposal = wiring["proposals"][0]
assert proposal["incident_id"] == "sonar/ariadne/python:S1172/AZ-1" assert proposal["incident_id"] == "sonar/ariadne/AZ-1"
assert proposal["job"] == "ariadne" assert proposal["job"] == "ariadne"
assert proposal["build_number"] == "sonar-AZ-1" assert proposal["build_number"] == "sonar-AZ-1"
@ -164,7 +159,7 @@ def test_the_bundle_offers_only_the_file_the_finding_names() -> None:
assert bundle["sonarqube"] == {"project": "ariadne", "issues": [_issue()]} assert bundle["sonarqube"] == {"project": "ariadne", "issues": [_issue()]}
assert bundle["jenkins"]["console_failures"] == [] assert bundle["jenkins"]["console_failures"] == []
assert bundle["jenkins"]["console_tail"] == "" assert bundle["jenkins"]["console_tail"] == ""
assert bundle["incident_id"] == "sonar/ariadne/python:S1172/AZ-1" assert bundle["incident_id"] == "sonar/ariadne/AZ-1"
def test_the_sweep_budget_stops_after_its_quota(wiring) -> None: def test_the_sweep_budget_stops_after_its_quota(wiring) -> None:
@ -187,13 +182,13 @@ def test_a_declined_proposal_is_reported_and_does_not_spend_budget(wiring) -> No
assert result["proposed"] == 0 assert result["proposed"] == 0
assert len(wiring["proposals"]) == 2 assert len(wiring["proposals"]) == 2
assert any("open_proposal_limit_reached" in item for item in result["skipped"]) assert "a: open_proposal_limit_reached" in result["skipped"]
def test_no_configured_projects_makes_no_call(wiring) -> None: def test_no_configured_projects_makes_no_call(wiring) -> None:
result = module.sweep(_Storage(), _config(hermes_sonar_projects={}), {}) result = module.sweep(_Storage(), _config(hermes_sonar_projects={}), {})
assert result == {"proposed": 0, "advised": 0, "skipped": ["no_projects_configured"], "projects": 0} assert result == {"proposed": 0, "skipped": ["no_projects_configured"], "projects": 0}
assert wiring["proposals"] == [] assert wiring["proposals"] == []
@ -216,16 +211,12 @@ def test_a_fetch_error_is_reported_not_raised(wiring) -> None:
assert result["skipped"] == ["ariadne: sonar fetch http 503"] assert result["skipped"] == ["ariadne: sonar fetch http 503"]
def test_a_finding_too_large_to_patch_becomes_advice(wiring) -> None: def test_a_project_with_nothing_mechanical_is_reported(wiring) -> None:
"""The bulk of the backlog is refactors no anchored patch can express."""
wiring["issues"] = [_issue(effort="8h")] wiring["issues"] = [_issue(effort="8h")]
result = module.sweep(_Storage(), _config(), {}) result = module.sweep(_Storage(), _config(), {})
assert result["proposed"] == 0 assert result["skipped"] == ["ariadne: no mechanically fixable finding"]
assert result["skipped"] == ["ariadne: too large to patch (advice disabled)"]
assert wiring["proposals"] == []
def test_a_sweep_never_takes_down_the_scheduler(monkeypatch) -> None: def test_a_sweep_never_takes_down_the_scheduler(monkeypatch) -> None:
@ -369,109 +360,3 @@ def test_a_bundles_origin_decides_the_framing(bundle, expected) -> None:
from ariadne.services import hermes_code_prompt from ariadne.services import hermes_code_prompt
assert hermes_code_prompt.is_quality_sweep(bundle) is expected assert hermes_code_prompt.is_quality_sweep(bundle) is expected
def test_a_rule_already_under_review_is_not_proposed_again(wiring) -> None:
"""One rule is one root cause; thirty near-identical PRs get read as none."""
wiring["open_incidents"] = ["sonar/ariadne/python:S1172/AZ-OTHER"]
wiring["issues"] = [
_issue(key="a", rule="python:S1172"),
_issue(key="b", rule="python:S2208", effort="10min"),
]
result = module.sweep(_Storage(), _config(), {})
assert result["proposed"] == 1
assert wiring["proposals"][0]["incident_id"] == "sonar/ariadne/python:S2208/b"
def test_every_rule_under_review_means_nothing_new_to_propose(wiring) -> None:
wiring["open_incidents"] = ["sonar/ariadne/python:S1172/AZ-OTHER"]
wiring["issues"] = [_issue(rule="python:S1172")]
result = module.sweep(_Storage(), _config(), {})
assert result["proposed"] == 0
assert result["skipped"] == ["ariadne: no new finding to act on"]
def test_another_projects_open_proposal_does_not_block_this_one(wiring) -> None:
wiring["open_incidents"] = ["sonar/metis/python:S1172/AZ-X", "ariadne/408"]
assert module.sweep(_Storage(), _config(), {})["proposed"] == 1
def test_the_rules_under_review_are_read_from_the_open_proposals(monkeypatch) -> None:
monkeypatch.setattr(
module.hermes_code_repair, "open_proposal_incidents",
lambda cfg: (
[
"sonar/ariadne/python:S2208/AZ1",
"sonar/ariadne/python:S3776/AZ2",
"sonar/other/python:S1172/AZ3",
"ariadne/408",
"sonar/ariadne/malformed",
],
None,
),
)
assert module.proposed_rules(REPO_CFG, "ariadne") == {"python:S2208", "python:S3776"}
def test_an_unreadable_proposal_list_costs_a_duplicate_not_a_silent_rule(monkeypatch) -> None:
"""Failing open here means one extra PR; failing closed loses a whole rule."""
monkeypatch.setattr(
module.hermes_code_repair, "open_proposal_incidents", lambda cfg: ([], "http 500")
)
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"

View File

@ -443,39 +443,3 @@ def test_retry_access_request_reports_update_failure(monkeypatch) -> None:
resp = client.post("/api/access/requests/REQ1/retry") resp = client.post("/api/access/requests/REQ1/retry")
assert resp.status_code == 502 assert resp.status_code == 502
def test_hermes_autotriage_can_be_run_on_demand(monkeypatch) -> None:
"""The cron tick is once a minute; a demo should not wait for it."""
ctx = AuthContext(username="bstein", email="", groups=["admin"], claims={})
client = _client(monkeypatch, ctx)
calls = []
monkeypatch.setattr(
app_module,
"run_hermes_autotriage",
lambda storage: calls.append(storage) or {"status": "ok", "jobs": {}},
)
admin_run = client.post(
"/api/admin/hermes/autotriage/run",
headers={"Authorization": "Bearer token"},
)
internal_run = client.post("/api/internal/hermes/autotriage/run")
assert admin_run.status_code == 200
assert internal_run.status_code == 200
assert admin_run.json()["status"] == "ok"
assert len(calls) == 2
def test_hermes_autotriage_on_demand_requires_admin(monkeypatch) -> None:
ctx = AuthContext(username="nobody", email="", groups=[], claims={})
client = _client(monkeypatch, ctx)
monkeypatch.setattr(app_module, "run_hermes_autotriage", lambda storage: {"status": "ok"})
resp = client.post(
"/api/admin/hermes/autotriage/run",
headers={"Authorization": "Bearer token"},
)
assert resp.status_code == 403