ariadne/ariadne/services/hermes_autotriage.py

from __future__ import annotations

import json
import time
from typing import Any

import httpx

from ..settings import settings
from ..utils.logging import get_logger
from . import hermes_agent_client, hermes_autotriage_repair, hermes_code_flow, hermes_incident_issue
from . import hermes_infra_signals
from . import hermes_autotriage_decision as hermes_decision
from . import hermes_autotriage_events as hermes_events
from . import hermes_autotriage_evidence as hermes_evidence
from .hermes_autotriage_metrics import (
    HERMES_TRIAGE_ACTION_TOTAL,
    HERMES_TRIAGE_DURATION_SECONDS,
    HERMES_TRIAGE_INCIDENT,
    HERMES_TRIAGE_LAST_SUCCESS_TS,
    KNOWN_ACTION_LABELS,
    refresh_incident_gauges,
)


logger = get_logger(__name__)

INCIDENT_EVENT_TYPE = hermes_events.INCIDENT_EVENT_TYPE
DIAGNOSIS_EVENT_TYPE = hermes_events.DIAGNOSIS_EVENT_TYPE
ACTION_EVENT_TYPE = hermes_events.ACTION_EVENT_TYPE
EXPECTED_CLASSIFICATION = "known_demo_fixture_failure"
REPAIR_ACTION = "repair_demo_fixture"
RETRY_ACTION = "retry_transient_infra"
REBUILD_FAILED_REASON = "repair rebuild failed"
CODE_FIX_PROPOSED_REASON = "code_fix_proposed"

_LAST_BUILD_TREE = "lastBuild[number,result,building,timestamp,duration,url]"
_RUN_COMPLETED = "completed"
_UNKNOWN_ACTION_LABEL = "unknown"

_PROMPT_TEMPLATE = """Use $triage-titan-test-failures.
Analyze incident __INCIDENT_ID__.
Treat the attached Ariadne bundle as the source of truth.
Identify the first enforced failure.
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": "<string; use known_demo_fixture_failure only when the evidence shows the hermes-triage-demo fixture unhealthy signature>", "confidence": <0..1>, "facts": [{"statement": "...", "source": "jenkins|opensearch|victoriametrics|kubernetes|flux|gitea", "reference": "..."}], "inferences": ["..."], "first_failed_gate": "<string>", "requested_action": {"type": "run_ariadne_job", "id": "repair_demo_fixture"} or null, "human_required": <bool>, "reason": "<string>"}
You are diagnosing only; you do not execute anything. Ariadne separately validates and executes the requested action under its own authorization policy.
Set human_required to false when the evidence matches the known demo fixture signature and the appropriate response is the predefined repair_demo_fixture action.
Set human_required to true only when the failure does not match a known signature, decisive evidence is missing, or no allowlisted action fits.
Use classification transient_infra_failure with requested_action {"type": "run_ariadne_job", "id": "retry_transient_infra"} only when the evidence shows an infrastructure, connectivity, or registry error unrelated to the repository's code or tests (DNS resolution failure, connection refused, reset, or timed out, TLS handshake failure, image pull failure, or a 5xx from a registry or SCM host), because re-running the same commit is then the whole remediation.
Otherwise leave requested_action null and set human_required per the rules above.
Do not perform mutations.

Bundle:
__BUNDLE__"""

def run_hermes_autotriage(storage: Any) -> dict[str, Any]:
    """Run one Hermes auto-triage tick over the allowlisted Jenkins jobs.

    Inputs: a storage object providing record_event/list_events for the
    incident event log. Outputs: a summary dict for scheduler logging;
    {"status": "disabled"} when the feature flag is off.
    """

    if not settings.hermes_autotriage_enabled:
        return {"status": "disabled"}
    started = time.time()
    incidents = hermes_events.incident_state(storage)
    jobs: dict[str, Any] = {}
    tick_state: dict[str, Any] = {}
    for job in settings.hermes_autotriage_job_allowlist:
        jobs[job] = _process_job(storage, job, incidents, tick_state)
    HERMES_TRIAGE_DURATION_SECONDS.labels(phase="total").set(time.time() - started)
    logger.info(
        "hermes autotriage tick finished",
        extra={"event": "hermes_autotriage", "status": "ok", "jobs": json.dumps(jobs, ensure_ascii=True)},
    )
    return {"status": "ok", "jobs": jobs}


def _process_job(
    storage: Any, job: str, incidents: dict[str, dict[str, Any]], tick_state: dict[str, Any]
) -> dict[str, Any]:
    """Inspect one allowlisted job's last build and advance its incidents."""

    last_build = _fetch_last_build(job)
    if last_build is None or last_build.get("building") or last_build.get("number") is None:
        return {"status": "skipped"}
    refresh_incident_gauges(job, last_build, incidents)
    result = str(last_build.get("result") or "").upper()
    if result == "SUCCESS":
        return _resolve_on_success(storage, job, last_build, incidents)
    if result == "FAILURE":
        return _handle_failure(storage, job, last_build, incidents, tick_state)
    return {"status": "ignored", "result": result}


def _fetch_last_build(job: str) -> dict[str, Any] | None:
    """Fetch the job's lastBuild summary from Jenkins, or None on failure."""

    base_url = settings.jenkins_base_url.strip().rstrip("/")
    if not base_url:
        return None
    try:
        with httpx.Client(**_jenkins_client_kwargs()) as client:
            response = client.get(f"{base_url}/job/{job}/api/json", params={"tree": _LAST_BUILD_TREE})
            response.raise_for_status()
            payload = response.json()
    except Exception as exc:
        logger.info(
            "hermes autotriage jenkins fetch failed",
            extra={"event": "hermes_autotriage", "status": "jenkins_error", "job": job, "detail": str(exc)},
        )
        return None
    last_build = payload.get("lastBuild") if isinstance(payload, dict) else None
    return last_build if isinstance(last_build, dict) else None


def _jenkins_client_kwargs() -> dict[str, Any]:
    """Build httpx client kwargs with Ariadne's Jenkins read credential."""

    kwargs: dict[str, Any] = {
        "timeout": settings.jenkins_api_timeout_sec,
        "follow_redirects": True,
    }
    username = settings.jenkins_api_user.strip()
    token = settings.jenkins_api_token.strip()
    if username and token:
        kwargs["auth"] = (username, token)
    return kwargs


def _resolve_on_success(
    storage: Any, job: str, last_build: dict[str, Any], incidents: dict[str, dict[str, Any]]
) -> dict[str, Any]:
    """Resolve incidents whose rebuild completed with this successful build."""

    number = _int_value(last_build.get("number"))
    resolved: list[str] = []
    for incident in incidents.values():
        if (
            incident.get("job") == job
            and incident.get("status") == "awaiting_rebuild"
            and _int_value(incident.get("build_number")) < number
        ):
            base = hermes_events.incident_base(incident)
            hermes_events.record_incident(storage, base, "resolved", {"resolved_by_build": number})
            HERMES_TRIAGE_LAST_SUCCESS_TS.set(time.time())
            resolved.append(str(base["incident_id"]))
    return {"status": "healthy", "resolved": resolved}


def _handle_failure(  # noqa: PLR0913 - the tick's issue budget travels with the incident context
    storage: Any, job: str, last_build: dict[str, Any], incidents: dict[str, dict], tick_state: dict[str, Any]
) -> dict[str, Any]:
    """Route a terminal build failure to dedupe, rebuild-failure, or triage."""

    number = _int_value(last_build.get("number"))
    incident_id = f"{job}/{number}"
    existing = incidents.get(incident_id)
    if existing is not None and existing.get("status") != "detected":
        return {"status": "deduped", "incident_id": incident_id}
    stale = _awaiting_rebuild_incident(incidents, job, incident_id)
    if stale is not None:
        base = {"incident_id": incident_id, "job": job, "build_number": number}
        return _mark_rebuild_failure(storage, stale, base)
    return _run_pipeline(storage, incident_id, job, last_build, tick_state)


def _awaiting_rebuild_incident(
    incidents: dict[str, dict[str, Any]], job: str, exclude_id: str
) -> dict[str, Any] | None:
    """Find a different incident for this job still awaiting its rebuild."""

    for incident_id, incident in incidents.items():
        if (
            incident_id != exclude_id
            and incident.get("job") == job
            and incident.get("status") == "awaiting_rebuild"
        ):
            return incident
    return None


def _mark_rebuild_failure(
    storage: Any, stale: dict[str, Any], base: dict[str, Any]
) -> dict[str, Any]:
    """Fail the incident whose rebuild broke and escalate the new failure."""

    stale_base = hermes_events.incident_base(stale)
    hermes_events.record_incident(
        storage,
        stale_base,
        "failed",
        {"reason": REBUILD_FAILED_REASON, "failed_rebuild": base["incident_id"]},
    )
    hermes_events.record_incident(storage, base, "human_required", {"reason": REBUILD_FAILED_REASON})
    return {
        "status": "rebuild_failed",
        "incident_id": str(base["incident_id"]),
        "failed_incident": str(stale_base["incident_id"]),
    }


def _run_pipeline(  # noqa: PLR0913 - the tick's issue budget travels with the incident context
    storage: Any, incident_id: str, job: str, last_build: dict[str, Any], tick_state: dict[str, Any]
) -> dict[str, Any]:
    """Run detect, evidence, diagnosis, and authorization for a new incident."""

    base = {"incident_id": incident_id, "job": job, "build_number": _int_value(last_build.get("number"))}
    hermes_events.record_incident(
        storage,
        base,
        "detected",
        {"result": str(last_build.get("result") or ""), "url": str(last_build.get("url") or "")},
    )
    phase_started = time.time()
    bundle = hermes_evidence.collect_evidence(incident_id, job, last_build)
    HERMES_TRIAGE_DURATION_SECONDS.labels(phase="evidence").set(time.time() - phase_started)
    if settings.hermes_code_enabled and job == settings.hermes_code_job:
        return _propose_code_fix(storage, base, bundle)
    phase_started = time.time()
    run = hermes_agent_client.run_triage(_hermes_run_config(), _build_prompt(incident_id, bundle))
    HERMES_TRIAGE_DURATION_SECONDS.labels(phase="diagnosis").set(time.time() - phase_started)
    if run.status != _RUN_COMPLETED or not run.output:
        reason = f"hermes_run_{run.status}"
        hermes_events.record_diagnosis(storage, base, run, None, hermes_events.Authorization(False, reason))
        hermes_events.record_incident(storage, base, "human_required", {"reason": reason})
        _file_incident_issue(storage, base, _diagnosis(bundle, None, reason, run.run_id), tick_state)
        return {"status": "human_required", "incident_id": incident_id, "reason": reason}
    outcome = hermes_decision.parse_triage_response(run.output, incident_id)
    return _authorize_and_execute(storage, base, run, outcome, bundle, tick_state)


def _authorize_and_execute(  # noqa: PLR0913 - the tick's issue budget travels with the incident context
    storage: Any, base: dict[str, Any], run: Any, outcome: Any, bundle: dict[str, Any], tick_state: dict[str, Any]
) -> dict[str, Any]:
    """Gate the parsed diagnosis and run its action when every gate passes.

    Escalating to a human is also where a mapped real service job earns a
    code-fix proposal: the pull request rides along on the same human_required
    phase and issue and never changes the outcome, while a job whose failure
    was auto-remediated gets no patch at all.
    """

    incident_id = str(base["incident_id"])
    matched, marker = _evidence_signature(outcome, bundle, incident_id)
    allowed, reason = hermes_decision.authorize_action(
        outcome,
        _decision_config(),
        hermes_events.prior_action_count(storage, incident_id),
        build_is_terminal_failure=True,
        job_allowlisted=True,
        evidence_has_signature=matched,
    )
    hermes_events.record_diagnosis(
        storage, base, run, outcome, hermes_events.Authorization(allowed, reason, marker)
    )
    if outcome.valid:
        hermes_events.record_incident(storage, base, "diagnosed", hermes_events.outcome_phase(outcome))
    if not allowed:
        HERMES_TRIAGE_ACTION_TOTAL.labels(action=_action_label(outcome), result="rejected").inc()
        proposal = hermes_code_flow.propose_for_incident(
            storage, base, bundle, _hermes_run_config(), settings
        )
        hermes_events.record_incident(storage, base, "human_required", {"reason": reason, **proposal})
        diagnosis = {**_diagnosis(bundle, outcome, reason, run.run_id), **proposal}
        _file_incident_issue(storage, base, diagnosis, tick_state)
        return {"status": "human_required", "incident_id": incident_id, "reason": reason}
    return _execute_action(storage, base, outcome, marker)


def _file_incident_issue(
    storage: Any, base: dict[str, Any], diagnosis: dict[str, Any], tick_state: dict[str, Any]
) -> None:
    """Hand a human_required incident to its service repository as an issue.

    Filing is opt-in per job, additive, and never mutates anything, so any
    failure is swallowed here and the tick continues unaffected.
    """

    try:
        hermes_incident_issue.maybe_file_issue(storage, settings, base, diagnosis, tick_state)
    except Exception as exc:
        logger.info(
            "hermes autotriage issue filing failed",
            extra={"event": "hermes_autotriage", "status": "issue_error", "detail": str(exc)},
        )


def _diagnosis(bundle: dict[str, Any], outcome: Any, reason: str, run_id: str | None) -> dict[str, Any]:
    """Pack the diagnosis inputs the issue body is rendered from."""

    return {"bundle": bundle, "outcome": outcome, "authorize_reason": reason, "run_id": run_id}


def _evidence_signature(outcome: Any, bundle: dict[str, Any], incident_id: str) -> tuple[bool, str | None]:
    """Run the signature check that belongs to the requested action id.

    Every allowlisted action has its own idea of decisive evidence: the
    fixture repair needs the demo failure signature, the transient retry
    needs an infrastructure marker. Returns (matched, marker) where marker
    names the console text that justified a retry, or None when the action's
    signature check does not name one.
    """

    if _requested_action_id(outcome) == RETRY_ACTION:
        return hermes_infra_signals.has_transient_infra_signature(bundle)
    return hermes_evidence.evidence_has_signature(bundle, incident_id), None


def _propose_code_fix(storage: Any, base: dict[str, Any], bundle: dict[str, Any]) -> dict[str, Any]:
    """Run the code-repair proposal flow and park the incident with humans.

    A pull request always awaits human review, so the incident lands in
    human_required either way; this path never touches the fixture-repair
    action accounting.
    """

    result = hermes_code_flow.propose_code_fix(
        storage,
        str(base["incident_id"]),
        str(base["job"]),
        _int_value(base["build_number"]),
        bundle,
        _hermes_run_config(),
        hermes_code_flow.code_config(settings),
    )
    if result.get("status") == "pr_opened":
        reason = CODE_FIX_PROPOSED_REASON
        phase = {
            "reason": reason,
            "branch": result.get("branch"),
            "pr_number": result.get("pr_number"),
            "url": result.get("url"),
        }
    else:
        reason = str(result.get("reason") or "code_fix_not_proposed")
        phase = {"reason": reason}
    hermes_events.record_incident(storage, base, "human_required", phase)
    return {"status": "human_required", "incident_id": str(base["incident_id"]), "reason": reason}


def _execute_action(
    storage: Any, base: dict[str, Any], outcome: Any, marker: str | None
) -> dict[str, Any]:
    """Dispatch the authorized action id to its registered executor."""

    action_id = _action_label(outcome)
    if action_id == RETRY_ACTION:
        return _retry_transient_infra(storage, base, marker)
    return _repair_demo_fixture(storage, base, action_id)


def _repair_demo_fixture(storage: Any, base: dict[str, Any], action_id: str) -> dict[str, Any]:
    """Run the demo-fixture repair Job and request the verification rebuild."""

    hermes_events.record_action(storage, base, action_id, "requested", None)
    hermes_events.record_action(storage, base, action_id, "accepted", None)
    hermes_events.record_incident(storage, base, "repairing", {"action": action_id})
    phase_started = time.time()
    repair = hermes_autotriage_repair.execute_repair(
        _repair_config(), str(base["incident_id"]), _int_value(base["build_number"])
    )
    HERMES_TRIAGE_DURATION_SECONDS.labels(phase="repair").set(time.time() - phase_started)
    if not repair.get("succeeded"):
        return _fail_action(storage, base, action_id, str(repair.get("error") or "repair failed"))
    rebuild = hermes_autotriage_repair.trigger_rebuild(settings, str(base["job"]))
    if not rebuild.get("requested"):
        return _fail_action(storage, base, action_id, str(rebuild.get("error") or "rebuild trigger failed"))
    hermes_events.record_action(storage, base, action_id, "executed", {"repair_job": repair.get("job_name")})
    hermes_events.record_incident(
        storage,
        base,
        "awaiting_rebuild",
        {"action": action_id, "repair_job": repair.get("job_name")},
    )
    return {
        "status": "awaiting_rebuild",
        "incident_id": str(base["incident_id"]),
        "repair_job": repair.get("job_name"),
    }


def _retry_transient_infra(storage: Any, base: dict[str, Any], marker: str | None) -> dict[str, Any]:
    """Re-run a build that failed for a demonstrably transient infra reason.

    No Kubernetes Job runs here: the rebuild is the whole remediation. The
    incident parks in awaiting_rebuild so the existing success-resolution
    logic closes it on the job's next green build, and the
    one-action-per-incident budget stops the retry from looping.
    """

    job = str(base["job"])
    detail = {"evidence_marker": marker}
    hermes_events.record_action(storage, base, RETRY_ACTION, "requested", detail)
    hermes_events.record_action(storage, base, RETRY_ACTION, "accepted", None)
    retry = hermes_autotriage_repair.retry_build(
        settings, job, parameterized=_is_parameterized_job(job)
    )
    if not retry.get("requested"):
        return _fail_action(storage, base, RETRY_ACTION, str(retry.get("error") or "retry trigger failed"))
    hermes_events.record_action(storage, base, RETRY_ACTION, "executed", detail)
    hermes_events.record_incident(storage, base, "awaiting_rebuild", {"action": RETRY_ACTION, **detail})
    return {
        "status": "awaiting_rebuild",
        "incident_id": str(base["incident_id"]),
        "action": RETRY_ACTION,
        "evidence_marker": marker,
    }


def _fail_action(storage: Any, base: dict[str, Any], action_id: str, error: str) -> dict[str, Any]:
    """Record a failed remediation and flag the incident for humans."""

    hermes_events.record_action(storage, base, action_id, "failed", {"error": error})
    hermes_events.record_incident(
        storage, base, "failed", {"reason": error}, extra_statuses=("human_required",)
    )
    return {"status": "failed", "incident_id": str(base["incident_id"]), "reason": error}


def _build_prompt(incident_id: str, bundle: dict[str, Any]) -> str:
    """Render the frozen triage prompt with the incident id and bundle."""

    compact = json.dumps(bundle, separators=(",", ":"), ensure_ascii=True)
    return _PROMPT_TEMPLATE.replace("__INCIDENT_ID__", incident_id).replace("__BUNDLE__", compact)


def _requested_action_id(outcome: Any) -> str:
    """Return the raw action id a triage response asked for, if any."""

    decision = outcome.decision if outcome is not None else None
    action = decision.requested_action if decision is not None else None
    return str(action.id) if action is not None else ""


def _action_label(outcome: Any) -> str:
    """Return a bounded metric label for the requested action id.

    Ids outside the action registry and the deployed allowlist collapse to
    "unknown" so a hallucinated action can never create a new metric series.
    """

    action_id = _requested_action_id(outcome)
    if action_id and (action_id in settings.hermes_allowed_actions or action_id in KNOWN_ACTION_LABELS):
        return action_id
    return _UNKNOWN_ACTION_LABEL


def _is_parameterized_job(job: str) -> bool:
    """Report whether a Jenkins job declares build parameters."""

    return job in list(settings.hermes_parameterized_jobs)


def _hermes_run_config() -> dict[str, Any]:
    """Build the config passed to the frozen Hermes agent client."""

    return {
        "base_url": settings.hermes_api_url,
        "api_key": settings.hermes_api_key,
        "total_timeout_seconds": settings.hermes_run_timeout_seconds,
    }


def _decision_config() -> dict[str, Any]:
    """Build the gate config passed to the frozen authorization chain."""

    return {
        "autoremediation_enabled": settings.hermes_autoremediation_enabled,
        "allowed_actions": list(settings.hermes_allowed_actions),
        "min_confidence": settings.hermes_min_confidence,
        "expected_classification": EXPECTED_CLASSIFICATION,
        "action_classifications": dict(settings.hermes_action_classifications),
        "max_actions_per_incident": settings.hermes_max_actions_per_incident,
    }


def _repair_config() -> dict[str, Any]:
    """Build the config passed to the repair Job executor."""

    return {
        "namespace": settings.hermes_demo_namespace,
        "fixture_configmap": settings.hermes_demo_fixture_configmap,
        "image": settings.hermes_repair_image,
    }


def _int_value(value: Any) -> int:
    """Coerce a value to int, defaulting to zero."""

    try:
        return int(value)
    except (TypeError, ValueError):
        return 0