ariadne/ariadne/services/hermes_autotriage_actions.py

"""Execute the actions Ariadne is allowed to take on its own authority.

Kept apart from the orchestrator because this is the only code in triage
that changes anything outside Ariadne, and it should be reviewable as one
unit. Every executor here corresponds to an id in the action allowlist;
nothing reaches these functions without passing the full gate chain.
"""

from __future__ import annotations

import time
from typing import Any

from ..settings import settings
from . import hermes_autotriage as hermes_autotriage
from . import hermes_autotriage_events as hermes_events
from . import hermes_autotriage_repair, jenkins_workspace_cleanup, pod_cleaner
from .hermes_autotriage_metrics import (
    HERMES_TRIAGE_ACTION_TOTAL,
    HERMES_TRIAGE_DURATION_SECONDS,
)


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 = hermes_autotriage._action_label(outcome)
    if action_id == hermes_autotriage.RETRY_ACTION:
        return _retry_transient_infra(storage, base, marker)
    if action_id == hermes_autotriage.RECLAIM_ACTION:
        return _reclaim_workspace_storage(storage, base, marker)
    if action_id == hermes_autotriage.AGENT_ACTION:
        return _clear_stuck_agent_pods(storage, base, marker)
    return _repair_demo_fixture(storage, base, action_id)


def _clear_stuck_agent_pods(storage: Any, base: dict[str, Any], marker: str | None) -> dict[str, Any]:
    """Clear finished agent pods, then request one rebuild.

    A build whose agent never started often succeeds on a retry, but when the
    pool is already full of stuck pods the retry queues behind them and fails
    identically. Clearing first is what makes the retry worth making. The
    clear is Ariadne's existing scheduled pod cleanup, which only removes pods
    that have already succeeded or failed, so nothing running is touched.
    """

    action = hermes_autotriage.AGENT_ACTION
    hermes_events.record_action(storage, base, action, "requested", {"marker": marker})
    try:
        summary = pod_cleaner.clean_finished_pods()
        cleared = getattr(summary, "deleted", 0)
    except Exception as exc:
        return fail_action(storage, base, action, f"pod cleanup failed: {exc}")
    hermes_events.record_action(storage, base, action, "executed", {"cleared": cleared})
    rebuild = hermes_autotriage_repair.trigger_rebuild(settings, str(base["job"]))
    if not rebuild.get("requested"):
        return fail_action(storage, base, action, str(rebuild.get("error") or "rebuild failed"))
    hermes_events.record_incident(storage, base, "awaiting_rebuild", {"action": action, "cleared": cleared})
    return {"status": "awaiting_rebuild", "incident_id": str(base["incident_id"]), "cleared": cleared}


def _reclaim_workspace_storage(storage: Any, base: dict[str, Any], marker: str | None) -> dict[str, Any]:
    """Reclaim stale Jenkins workspace storage, then request one rebuild.

    A rebuild alone lands on the same full volume, which is why disk
    exhaustion is excluded from the transient retry. Reclaiming first makes
    the retry meaningful. The reclaim is Ariadne's existing scheduled cleanup,
    with its own deletion budget, so this grants no new capability.
    """

    hermes_events.record_action(storage, base, hermes_autotriage.RECLAIM_ACTION, "requested", {"marker": marker})
    try:
        summary = jenkins_workspace_cleanup.cleanup_jenkins_workspace_storage()
        freed = getattr(summary, "removed_pvcs", 0)
    except Exception as exc:
        return fail_action(storage, base, hermes_autotriage.RECLAIM_ACTION, f"workspace reclaim failed: {exc}")
    hermes_events.record_action(storage, base, hermes_autotriage.RECLAIM_ACTION, "executed", {"reclaimed": freed})
    rebuild = hermes_autotriage_repair.trigger_rebuild(settings, str(base["job"]))
    if not rebuild.get("requested"):
        return fail_action(storage, base, hermes_autotriage.RECLAIM_ACTION, str(rebuild.get("error") or "rebuild failed"))
    hermes_events.record_incident(
        storage, base, "awaiting_rebuild", {"action": hermes_autotriage.RECLAIM_ACTION, "reclaimed": freed}
    )
    return {"status": "awaiting_rebuild", "incident_id": str(base["incident_id"]), "reclaimed": freed}


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(
        hermes_autotriage._repair_config(), str(base["incident_id"]), hermes_autotriage._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, hermes_autotriage.RETRY_ACTION, "requested", detail)
    hermes_events.record_action(storage, base, hermes_autotriage.RETRY_ACTION, "accepted", None)
    retry = hermes_autotriage_repair.retry_build(
        settings, job, parameterized=hermes_autotriage._is_parameterized_job(job)
    )
    if not retry.get("requested"):
        return fail_action(storage, base, hermes_autotriage.RETRY_ACTION, str(retry.get("error") or "retry trigger failed"))
    hermes_events.record_action(storage, base, hermes_autotriage.RETRY_ACTION, "executed", detail)
    hermes_events.record_incident(storage, base, "awaiting_rebuild", {"action": hermes_autotriage.RETRY_ACTION, **detail})
    return {
        "status": "awaiting_rebuild",
        "incident_id": str(base["incident_id"]),
        "action": hermes_autotriage.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}