feat(hermes): 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 in the write allowlist, and the change is one anchored snippet the
validator can check. When any of that fails the incident escalates with a
diagnosis and nothing else - even though the model that wrote the diagnosis
frequently knows exactly what the fix is. That knowledge was discarded at the
moment it was most useful, because the cases where no patch is possible are
exactly the cases a maintainer has to do by hand.

A diagnosis may now carry up to three code suggestions: the file, what is
wrong there, and the code to change it to. Rendered into the issue under a
heading that says the change was not applied, because a code block in an issue
reads like something that already happened unless it is told otherwise.

Deliberately not a patch, and the difference is the safety story. A patch must
survive the validator because Ariadne acts on it. A suggestion is read by a
person who is already going to edit that file, so being wrong costs them a
moment's thought rather than a bad commit - which is why suggestions may
describe changes too large or too diffuse for the patcher to have attempted,
and why nothing here is anchored, applied or pushed.

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.

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

ariadne/services/hermes_autotriage_decision.py

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

ariadne/services/hermes_autotriage_events.py

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

ariadne/services/hermes_code_suggestion.py

@@ -0,0 +1,157 @@
+"""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

ariadne/services/hermes_incident_body.py

@@ -15,6 +15,7 @@ from __future__ import annotations
 import re
 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_suggested_remediation as suggestion_field
 
@@ -122,6 +123,7 @@ def issue_body(context: dict, max_chars: int = DEFAULT_MAX_BODY_CHARS) -> str:
         _facts_section(context),
         evidence_section.evidence_section(context.get("bundle")),
         _inferences_section(context),
+        code_suggestion_field.issue_section(context.get("code_suggestions")),
         suggestion_field.issue_section(context.get("suggested_remediation")),
         _links_section(context),
         _footer(context),

ariadne/services/hermes_incident_issue.py

@@ -203,6 +203,7 @@ def issue_context(base: dict[str, Any], diagnosis: dict[str, Any]) -> dict[str,
         "facts": [body.fact_fields(fact) for fact in getattr(decision, "facts", None) or []],
         "inferences": list(getattr(decision, "inferences", None) or []),
         "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
         # where in Jenkins the failure can be found.
         "bundle": bundle,

ariadne/services/hermes_triage_prompt.py

@@ -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.
 Distinguish facts from inference.
 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>, "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>, "code_suggestions": <array 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.
 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.
@@ -35,6 +35,7 @@ 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.
 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 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.
 Do not perform mutations."""
 

tests/test_hermes_code_suggestion.py

@@ -0,0 +1,174 @@
+"""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