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_triage_prompt
from . import hermes_hung_builds
from . import hermes_jenkins_client
from . import hermes_multibranch
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"

_RUN_COMPLETED = "completed"
_UNKNOWN_ACTION_LABEL = "unknown"


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)
        for branch in jobs[job].get("branches") or []:
            jobs[branch] = _process_job(storage, branch, 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."""

    payload = hermes_jenkins_client.fetch_job_payload(job)
    if hermes_multibranch.is_folder(payload):
        branches = hermes_multibranch.branch_job_paths(payload, job, settings.hermes_max_branches)
        return {"status": "folder", "branches": branches}
    last_build = hermes_jenkins_client.last_build(payload)
    if last_build is None or last_build.get("number") is None:
        return {"status": "skipped"}
    if last_build.get("building"):
        return _handle_running_build(storage, job, last_build, incidents, tick_state)
    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 _handle_running_build(
    storage: Any, job: str, last_build: dict[str, Any], incidents: dict[str, dict], tick_state: dict[str, Any]
) -> dict[str, Any]:
    """Escalate a build that has run past the cap; otherwise leave it alone.

    A build that never terminates produces no incident anywhere while holding
    an agent slot. No model is consulted: its console is still being written,
    so any root cause would be invented.
    """

    cap = float(settings.hermes_hung_build_minutes)
    if not hermes_hung_builds.is_hung(last_build, cap):
        return {"status": "skipped"}
    incident_id = f"{job}/{_int_value(last_build.get('number'))}"
    if incidents.get(incident_id) is not None:
        return {"status": "deduped", "incident_id": incident_id}
    reason = hermes_hung_builds.HUNG_REASON
    base = {"incident_id": incident_id, "job": job, "build_number": _int_value(last_build.get("number"))}
    hermes_events.record_incident(storage, base, "human_required", {"reason": reason})
    bundle = hermes_hung_builds.hung_bundle(job, last_build, cap)
    hung = {"classification": hermes_hung_builds.HUNG_CLASSIFICATION}
    _file_incident_issue(storage, base, {**_diagnosis(bundle, None, reason, None), **hung}, tick_state)
    return {"status": "human_required", "incident_id": incident_id, "reason": reason}


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":
        _backfill_issue(storage, job, number, incident_id, existing, tick_state)
        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 _backfill_issue(  # noqa: PLR0913 - the tick's issue budget travels with the incident
    storage: Any, job: str, number: int, incident_id: str, existing: dict, tick_state: dict[str, Any]
) -> None:
    """File the issue an escalated incident never received.

    An incident triaged before issue filing existed, or while its repository
    was unmapped, is deduped on every later tick and would otherwise never
    reach the service's tracker even though the job is still red.

    Reuses the classification the original diagnosis recorded so the existing
    dedupe applies: one open issue per job and classification, refiled only if
    it is closed while the job still fails. Skipped when no classification was
    recorded, because filing under a guessed label would duplicate rather than
    match. Additive and never raises.
    """

    if existing.get("status") != "human_required":
        return
    classification = hermes_events.recorded_classification(storage, incident_id)
    if not classification:
        return
    base = {"incident_id": incident_id, "job": job, "build_number": number}
    bundle = {"jenkins": {"job": job, "build_number": number, "console_failures": [], "console_tail": ""}}
    diagnosis = {**_diagnosis(bundle, None, "issue_backfill", None), "classification": classification}
    _file_incident_issue(storage, base, diagnosis, 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(), hermes_triage_prompt.build_prompt(incident_id, job, 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]:
    """Patch the demo fixture healthy and request the verification rebuild.

    The repair runs in-process against the cluster API rather than as a
    Kubernetes Job, so the executed action carries the patched target instead
    of a Job name; the one-action-per-incident budget upstream is still what
    stops the repair from repeating.
    """

    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"))
    detail = {"repair": repair.get("action"), "target": repair.get("target")}
    hermes_events.record_action(storage, base, action_id, "executed", detail)
    hermes_events.record_incident(storage, base, "awaiting_rebuild", {"action": action_id, **detail})
    return {"status": "awaiting_rebuild", "incident_id": str(base["incident_id"]), **detail}


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 _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 naming the ConfigMap the repair action patches."""

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


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

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