ariadne/ariadne/services/hermes_incident_body.py

"""Render the title, body, and dedupe marker of a Hermes triage issue.

Pure formatting: nothing here talks to Gitea, reads settings, or raises. The
body is what an operator actually reads, so every list is capped and every
statement clipped — a diagnosis must fit in an issue without burying the one
line that says why a person is needed.

The hidden marker comment is the whole dedupe mechanism. A repeatedly failing
job opens a fresh incident per build; the marker is what lets the next build
recognize the issue already filed for the same job and classification.
"""

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


DEFAULT_MAX_BODY_CHARS = 8000
UNDIAGNOSED = "undiagnosed"

_MAX_TITLE_CHARS = 120
_MAX_FACTS = 10
_MAX_FACT_CHARS = 300
_MAX_INFERENCES = 6
_ELLIPSIS = "..."
_FACT_KEYS = ("statement", "source", "reference")
_TRUNCATION_NOTICE = (
    "\n\n_Truncated by Ariadne: the diagnosis exceeded the configured issue body limit. "
    "The complete evidence bundle is in the Ariadne audit trail._"
)
_AUDIT_NOTE = (
    "Full evidence bundle and audit trail live in Ariadne at `/api/admin/audit/events`, "
    "event types `hermes_autotriage_incident` and `hermes_autotriage_diagnosis`."
)
_FOOTER_TEMPLATE = (
    "Filed automatically by Ariadne from a Hermes Agent diagnosis ({run}). "
    "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
# finished console to diagnose. Claiming a diagnosis that did not happen would
# misrepresent where the conclusion came from.
_NO_RUN_FOOTER = (
    "Filed automatically by Ariadne. No Hermes diagnosis was requested for this incident: "
    "the finding is a direct observation, not a model conclusion. "
    "Hermes has no write access to this repository; no files or infrastructure were changed."
)
_MARKER_PATTERN = re.compile(
    r"<!--\s*hermes-triage\s+job=(?P<job>.*?)\s+classification=(?P<classification>.*?)"
    r"\s+incident=(?P<incident>.*?)\s*-->"
)


def issue_marker(job: str, classification: str, incident_id: str) -> str:
    """Render the hidden marker comment that identifies a filed issue.

    Inputs: the Jenkins job, the diagnosis classification, and the incident id.
    Outputs: a single-line HTML comment safe to embed in an issue body; values
    are flattened so a stray newline or comment terminator cannot break the
    marker or the markdown around it.
    """

    return (
        f"<!-- hermes-triage job={_marker_value(job)}"
        f" classification={_marker_value(classification)}"
        f" incident={_marker_value(incident_id)} -->"
    )


def parse_issue_marker(body: Any) -> dict[str, str] | None:
    """Extract the hermes-triage marker fields from an issue body.

    Inputs: an issue body, which may be missing or not a string. Outputs:
    {"job", "classification", "incident"} for the first marker found, or None
    when the body carries no marker.
    """

    if not isinstance(body, str):
        return None
    match = _MARKER_PATTERN.search(body)
    if match is None:
        return None
    return {name: match.group(name).strip() for name in ("job", "classification", "incident")}


def issue_title(context: dict) -> str:
    """Render the issue title, bounded so issue lists stay readable.

    Inputs: the issue context (job, build_number, classification). Outputs:
    `[hermes] {job} #{build}: {classification}` clipped to 120 characters.
    """

    prefix = f"[hermes] {context.get('job')} #{context.get('build_number')}: "
    classification = str(context.get("classification") or UNDIAGNOSED)
    room = _MAX_TITLE_CHARS - len(prefix)
    if room <= len(_ELLIPSIS):
        return f"{prefix}{classification}"[:_MAX_TITLE_CHARS]
    return f"{prefix}{_clip(classification, room)}"


def issue_body(context: dict, max_chars: int = DEFAULT_MAX_BODY_CHARS) -> str:
    """Render the markdown issue body with the hidden marker line last.

    Inputs: the issue context (incident identity, classification, confidence,
    first_failed_gate, reason, authorize_reason, facts, inferences, build_url,
    run_id) and the whole-body character cap. Outputs: the markdown body,
    truncated with an explicit notice when it would exceed the cap, always
    ending in the dedupe marker so truncation can never drop it.
    """

    marker = issue_marker(
        str(context.get("job") or ""),
        str(context.get("classification") or UNDIAGNOSED),
        str(context.get("incident_id") or ""),
    )
    sections = [
        _summary_line(context),
        _human_section(context),
        _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),
    ]
    body = "\n\n".join(section for section in sections if section)
    return f"{_bounded_body(body, max_chars - len(marker) - 2)}\n\n{marker}"


def fact_fields(fact: Any) -> dict[str, str]:
    """Normalize one cited fact from a triage dataclass or a plain dict.

    Inputs: a TriageFact or a {statement, source, reference} mapping. Outputs:
    the three fields as strings, empty when absent.
    """

    if isinstance(fact, dict):
        return {key: str(fact.get(key) or "") for key in _FACT_KEYS}
    return {key: str(getattr(fact, key, "") or "") for key in _FACT_KEYS}


def _summary_line(context: dict) -> str:
    """Render the one-line summary that opens the issue body."""

    incident_id = context.get("incident_id")
    classification = context.get("classification")
    run_id = str(context.get("run_id") or "")
    if not run_id or run_id == "unknown":
        # The headline is the first thing read, so it must not credit a model
        # for a finding Ariadne made on its own.
        return f"Ariadne recorded incident `{incident_id}` as **{classification}**."
    confidence = context.get("confidence")
    return (
        f"Hermes auto-triage classified incident `{incident_id}` as "
        f"**{classification}** (confidence {'n/a' if confidence is None else confidence}); "
        f"first failed gate: `{context.get('first_failed_gate') or 'unknown'}`."
    )


def _footer(context: dict) -> str:
    """Attribute the finding to a Hermes run, or to Ariadne when there was none."""

    run_id = str(context.get("run_id") or "")
    if not run_id or run_id == "unknown":
        return _NO_RUN_FOOTER
    # A bare id is something to copy; a link is a page to open, and that page
    # 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:
    """Render the section explaining why the incident needs a person."""

    lines = ["## Why a human is needed", str(context.get("reason") or "no reason recorded")]
    observation = str(context.get("observation") or "").strip()
    if observation:
        # Escalations that never reached a model carry their explanation here
        # rather than in facts, which are only populated from a diagnosis.
        lines.append(observation)
    authorize_reason = str(context.get("authorize_reason") or "")
    if authorize_reason:
        lines.append(f"Ariadne did not authorize automated remediation: `{authorize_reason}`.")
    return "\n\n".join(lines)


def _facts_section(context: dict) -> str:
    """Render the bounded evidence list cited by the diagnosis."""

    facts = context.get("facts")
    if not isinstance(facts, list) or not facts:
        return ""
    lines = ["## Facts"]
    for fact in facts[:_MAX_FACTS]:
        fields = fact_fields(fact)
        statement = _clip(fields["statement"], _MAX_FACT_CHARS)
        lines.append(f"- **{fields['source'] or 'unknown'}** — {statement} (`{fields['reference']}`)")
    return "\n".join(lines)


def _inferences_section(context: dict) -> str:
    """Render the bounded inference list from the diagnosis."""

    inferences = context.get("inferences")
    if not isinstance(inferences, list) or not inferences:
        return ""
    lines = ["## Inferences"]
    lines.extend(f"- {_clip(str(item), _MAX_FACT_CHARS)}" for item in inferences[:_MAX_INFERENCES])
    return "\n".join(lines)


def _links_section(context: dict) -> str:
    """Render the links pointing back at Jenkins, the proposal, and the audit log."""

    build_url = str(context.get("build_url") or "")
    proposal_url = str(context.get("code_proposal_url") or "")
    lines = [
        "## Links",
        f"- Failed build: {build_url}" if build_url else "- Failed build: url unavailable",
    ]
    if proposal_url:
        lines.append(f"- Proposed fix awaiting review: {proposal_url}")
    lines.append(f"- {_AUDIT_NOTE}")
    return "\n".join(lines)


def _bounded_body(body: str, budget: int) -> str:
    """Clip the rendered body to its budget with an explicit notice."""

    if budget <= 0:
        return ""
    if len(body) <= budget:
        return body
    room = budget - len(_TRUNCATION_NOTICE)
    return body[:room] + _TRUNCATION_NOTICE if room > 0 else body[:budget]


def _clip(value: str, limit: int) -> str:
    """Clip one string to a limit, marking that it was shortened."""

    if len(value) <= limit:
        return value
    return value[: limit - len(_ELLIPSIS)] + _ELLIPSIS


def _marker_value(value: str) -> str:
    """Flatten one marker field so it cannot break the comment or the body."""

    return " ".join(str(value).replace("-->", "").split()) or "unknown"