ariadne/tests/test_hermes_autotriage.py

from __future__ import annotations

import pytest

from ariadne.services import hermes_autotriage as module
from tests.hermes_autotriage_harness import (
    INCIDENT_ID,
    JOB,
    _build,
    _counter,
    _events,
    _gauge,
    _model_output,
    _prepare,
    _run,
    _seed_incident,
    _settings,
    _statuses,
)


def test_disabled_tick(monkeypatch) -> None:
    env = _prepare(monkeypatch, cfg=_settings(hermes_autotriage_enabled=False))
    assert module.run_hermes_autotriage(env.storage) == {"status": "disabled"}
    assert env.storage.events == []
    assert env.calls["gets"] == []


def test_healthy_tick_without_incidents(monkeypatch) -> None:
    env = _prepare(monkeypatch, last_build=_build(13, "SUCCESS"))
    summary = module.run_hermes_autotriage(env.storage)
    assert summary["status"] == "ok"
    assert summary["jobs"][JOB] == {"status": "healthy", "resolved": []}
    assert env.storage.events == []
    assert env.calls["triage"] == []


def test_success_resolves_only_older_awaiting_rebuild(monkeypatch) -> None:
    env = _prepare(monkeypatch, last_build=_build(13, "SUCCESS"))
    _seed_incident(env.storage, "awaiting_rebuild", build_number=12)
    _seed_incident(env.storage, "awaiting_rebuild", build_number=13)
    summary = module.run_hermes_autotriage(env.storage)
    assert summary["jobs"][JOB]["resolved"] == [INCIDENT_ID]
    resolved = _events(env.storage, module.INCIDENT_EVENT_TYPE)[-1]
    assert resolved["incident_id"] == INCIDENT_ID
    assert resolved["status"] == "resolved"
    assert resolved["phase"] == {"resolved_by_build": 13}
    assert _gauge("12", "resolved") == 1.0
    assert _gauge("12", "human_required") == 0.0
    assert module.HERMES_TRIAGE_LAST_SUCCESS_TS._value.get() > 0


def test_new_failure_full_happy_path(monkeypatch) -> None:
    success_before = _counter("repair_demo_fixture", "success")
    env = _prepare(monkeypatch)
    summary = module.run_hermes_autotriage(env.storage)
    job_summary = summary["jobs"][JOB]
    assert job_summary["status"] == "awaiting_rebuild"
    assert job_summary["repair"] == "configmap_patch"
    assert job_summary["target"] == "hermes-triage-demo/hermes-triage-demo-fixture"
    assert _statuses(env.storage) == ["detected", "diagnosed", "repairing", "awaiting_rebuild"]
    actions = _events(env.storage, module.ACTION_EVENT_TYPE)
    assert [action["result"] for action in actions] == ["requested", "accepted", "executed"]
    assert all(action["action"] == "repair_demo_fixture" for action in actions)
    diagnosis = _events(env.storage, module.DIAGNOSIS_EVENT_TYPE)[0]
    assert diagnosis["authorized"] is True
    assert diagnosis["authorize_reason"] == "authorized"
    assert diagnosis["run"] == {
        "status": "completed",
        "run_id": "run-1",
        "session_id": "sess-1",
        "error": None,
        "duration_seconds": 1.5,
        "denied_approvals": 0,
    }
    assert diagnosis["outcome"]["classification"] == "known_demo_fixture_failure"
    assert env.calls["repairs"] == [
        (
            {
                "namespace": "hermes-triage-demo",
                "fixture_configmap": "hermes-triage-demo-fixture",
            },
            INCIDENT_ID,
            12,
        )
    ]
    assert actions[-1]["detail"] == {
        "repair": "configmap_patch",
        "target": "hermes-triage-demo/hermes-triage-demo-fixture",
    }
    awaiting = _events(env.storage, module.INCIDENT_EVENT_TYPE)[-1]
    assert awaiting["phase"] == {
        "action": "repair_demo_fixture",
        "repair": "configmap_patch",
        "target": "hermes-triage-demo/hermes-triage-demo-fixture",
    }
    assert env.calls["rebuilds"] == [JOB]
    assert _counter("repair_demo_fixture", "success") == success_before + 1.0
    assert _gauge("12", "awaiting_rebuild") == 1.0
    assert _gauge("12", "detected") == 0.0
    for phase in ("evidence", "diagnosis", "repair", "total"):
        assert module.HERMES_TRIAGE_DURATION_SECONDS.labels(phase=phase)._value.get() >= 0.0


def test_prompt_is_frozen_shape(monkeypatch) -> None:
    env = _prepare(monkeypatch)
    module.run_hermes_autotriage(env.storage)
    config, prompt = env.calls["triage"][0]
    assert config == {
        "base_url": "http://hermes:8642",
        "api_key": "key",
        "total_timeout_seconds": 420.0,
    }
    assert prompt.startswith("Use $triage-titan-test-failures.\n")
    assert f"Analyze incident {INCIDENT_ID}." in prompt
    assert f'"<must equal {INCIDENT_ID}>"' in prompt
    assert "You are diagnosing only; you do not execute anything." in prompt
    assert "Set human_required to false when the evidence matches" in prompt
    assert "Do not perform mutations.\n\nBundle:\n" in prompt
    assert prompt.rstrip().endswith('"log_evidence":{"records":[]}}')


def test_observe_mode_requires_human_without_actions(monkeypatch) -> None:
    rejected_before = _counter("repair_demo_fixture", "rejected")
    env = _prepare(monkeypatch, cfg=_settings(hermes_autoremediation_enabled=False))
    summary = module.run_hermes_autotriage(env.storage)
    assert summary["jobs"][JOB] == {
        "status": "human_required",
        "incident_id": INCIDENT_ID,
        "reason": "autoremediation_disabled",
    }
    assert _statuses(env.storage) == ["detected", "diagnosed", "human_required"]
    assert _events(env.storage, module.ACTION_EVENT_TYPE) == []
    assert env.calls["repairs"] == []
    assert env.calls["rebuilds"] == []
    assert _counter("repair_demo_fixture", "rejected") == rejected_before + 1.0


def test_known_incident_is_deduped(monkeypatch) -> None:
    env = _prepare(monkeypatch)
    _seed_incident(env.storage, "human_required")
    summary = module.run_hermes_autotriage(env.storage)
    assert summary["jobs"][JOB] == {"status": "deduped", "incident_id": INCIDENT_ID}
    assert len(env.storage.events) == 1
    assert env.calls["triage"] == []


def test_incident_state_reads_json_string_detail(monkeypatch) -> None:
    env = _prepare(monkeypatch)
    _seed_incident(env.storage, "resolved", as_json=True)
    summary = module.run_hermes_autotriage(env.storage)
    assert summary["jobs"][JOB]["status"] == "deduped"


def test_failed_rebuild_marks_both_incidents(monkeypatch) -> None:
    env = _prepare(monkeypatch, last_build=_build(13, "FAILURE"))
    _seed_incident(env.storage, "awaiting_rebuild", build_number=12)
    summary = module.run_hermes_autotriage(env.storage)
    assert summary["jobs"][JOB] == {
        "status": "rebuild_failed",
        "incident_id": f"{JOB}/13",
        "failed_incident": INCIDENT_ID,
    }
    details = _events(env.storage, module.INCIDENT_EVENT_TYPE)[1:]
    assert [(d["incident_id"], d["status"]) for d in details] == [
        (INCIDENT_ID, "failed"),
        (f"{JOB}/13", "human_required"),
    ]
    assert details[1]["phase"] == {"reason": "repair rebuild failed"}
    assert env.calls["triage"] == []
    assert env.calls["repairs"] == []


@pytest.mark.parametrize("status", ["timeout", "lost", "error", "failed", "cancelled"])
def test_unfinished_hermes_run_requires_human(monkeypatch, status) -> None:
    env = _prepare(monkeypatch, run=_run(status=status, error="boom"))
    summary = module.run_hermes_autotriage(env.storage)
    assert summary["jobs"][JOB]["reason"] == f"hermes_run_{status}"
    assert _statuses(env.storage) == ["detected", "human_required"]
    diagnosis = _events(env.storage, module.DIAGNOSIS_EVENT_TYPE)[0]
    assert diagnosis["run"]["status"] == status
    assert diagnosis["outcome"] is None
    assert diagnosis["authorized"] is False
    assert env.calls["repairs"] == []


def test_invalid_response_requires_human(monkeypatch) -> None:
    rejected_before = _counter("unknown", "rejected")
    env = _prepare(monkeypatch, run=_run(output="no json here"))
    summary = module.run_hermes_autotriage(env.storage)
    assert summary["jobs"][JOB]["reason"].startswith("response_invalid")
    assert _statuses(env.storage) == ["detected", "human_required"]
    assert _counter("unknown", "rejected") == rejected_before + 1.0


def test_model_human_required_is_rejected(monkeypatch) -> None:
    env = _prepare(monkeypatch, run=_run(output=_model_output(human_required=True)))
    summary = module.run_hermes_autotriage(env.storage)
    assert summary["jobs"][JOB]["reason"] == "human_required"
    assert _statuses(env.storage) == ["detected", "diagnosed", "human_required"]
    assert env.calls["repairs"] == []


def test_missing_signature_is_rejected(monkeypatch) -> None:
    env = _prepare(monkeypatch, signature=False)
    summary = module.run_hermes_autotriage(env.storage)
    assert summary["jobs"][JOB]["reason"] == "evidence_signature_missing"
    assert _statuses(env.storage) == ["detected", "diagnosed", "human_required"]
    assert env.calls["repairs"] == []


def test_non_allowlisted_action_is_rejected(monkeypatch) -> None:
    rejected_before = _counter("unknown", "rejected")
    output = _model_output(requested_action={"type": "run_ariadne_job", "id": "other_action"})
    env = _prepare(monkeypatch, run=_run(output=output))
    summary = module.run_hermes_autotriage(env.storage)
    assert summary["jobs"][JOB]["reason"].startswith("action_not_allowlisted")
    assert _counter("unknown", "rejected") == rejected_before + 1.0


def test_prior_action_blocks_second_action(monkeypatch) -> None:
    env = _prepare(monkeypatch)
    _seed_incident(env.storage, "detected")
    env.storage.record_event(
        module.ACTION_EVENT_TYPE,
        {"incident_id": INCIDENT_ID, "action": "repair_demo_fixture", "result": "requested"},
    )
    summary = module.run_hermes_autotriage(env.storage)
    assert summary["jobs"][JOB]["reason"] == "max_actions_reached"
    assert env.calls["repairs"] == []


def test_repair_failure_marks_failed_and_human(monkeypatch) -> None:
    failed_before = _counter("repair_demo_fixture", "failed")
    env = _prepare(
        monkeypatch,
        repair={
            "action": "configmap_patch",
            "target": "hermes-triage-demo/hermes-triage-demo-fixture",
            "succeeded": False,
            "error": "configmap patch http 403",
        },
    )
    summary = module.run_hermes_autotriage(env.storage)
    assert summary["jobs"][JOB] == {
        "status": "failed",
        "incident_id": INCIDENT_ID,
        "reason": "configmap patch http 403",
    }
    assert _statuses(env.storage) == ["detected", "diagnosed", "repairing", "failed"]
    actions = _events(env.storage, module.ACTION_EVENT_TYPE)
    assert [action["result"] for action in actions] == ["requested", "accepted", "failed"]
    assert env.calls["rebuilds"] == []
    assert _counter("repair_demo_fixture", "failed") == failed_before + 1.0
    assert _gauge("12", "failed") == 1.0
    assert _gauge("12", "human_required") == 1.0


def test_rebuild_trigger_failure_marks_failed(monkeypatch) -> None:
    env = _prepare(monkeypatch, rebuild={"requested": False, "error": "rebuild http 500"})
    summary = module.run_hermes_autotriage(env.storage)
    assert summary["jobs"][JOB]["status"] == "failed"
    assert summary["jobs"][JOB]["reason"] == "rebuild http 500"
    actions = _events(env.storage, module.ACTION_EVENT_TYPE)
    assert [action["result"] for action in actions] == ["requested", "accepted", "failed"]


def test_jenkins_fetch_failure_skips_job(monkeypatch) -> None:
    env = _prepare(monkeypatch, jenkins_exc=RuntimeError("boom"))
    summary = module.run_hermes_autotriage(env.storage)
    assert summary["jobs"][JOB] == {"status": "skipped"}
    assert env.storage.events == []


def test_building_build_is_skipped(monkeypatch) -> None:
    env = _prepare(monkeypatch, last_build=_build(12, None, building=True))
    assert module.run_hermes_autotriage(env.storage)["jobs"][JOB] == {"status": "skipped"}
    assert env.storage.events == []


def test_empty_jenkins_base_url_skips(monkeypatch) -> None:
    env = _prepare(monkeypatch, cfg=_settings(jenkins_base_url=""))
    assert module.run_hermes_autotriage(env.storage)["jobs"][JOB] == {"status": "skipped"}
    assert env.calls["gets"] == []


def test_non_terminal_result_is_ignored(monkeypatch) -> None:
    env = _prepare(monkeypatch, last_build=_build(12, "ABORTED"))
    assert module.run_hermes_autotriage(env.storage)["jobs"][JOB] == {
        "status": "ignored",
        "result": "ABORTED",
    }
    assert env.storage.events == []


def test_event_detail_tolerates_bad_payloads() -> None:
    assert module.hermes_events.event_detail({"detail": "not-json"}) is None
    assert module.hermes_events.event_detail({"detail": "[1,2]"}) is None
    assert module.hermes_events.event_detail({"detail": 5}) is None
    assert module.hermes_events.event_detail("not-a-row") is None
    assert module._int_value("not-a-number") == 0


def test_jenkins_request_uses_basic_auth_and_tree(monkeypatch) -> None:
    env = _prepare(monkeypatch, last_build=_build(13, "SUCCESS"))
    module.run_hermes_autotriage(env.storage)
    assert env.calls["client_kwargs"]["auth"] == ("user", "token")
    url, params = env.calls["gets"][0]
    assert url == f"https://ci.example/job/{JOB}/api/json"
    assert params == {"tree": "lastBuild[number,result,building,timestamp,duration,url]"}


def _issue_settings(**overrides):  # type: ignore[no-untyped-def]
    values = {
        "hermes_issues_enabled": True,
        "hermes_issue_repos": {JOB: ("bstein", JOB)},
    }
    values.update(overrides)
    return _settings(**values)


def _install_issue_tracker(monkeypatch, calls) -> None:  # type: ignore[no-untyped-def]
    filed: dict = {}

    def fake_find(cfg, job, classification, incident_id):  # type: ignore[no-untyped-def]
        calls["lookups"].append((cfg["owner"], cfg["repo"], job, classification, incident_id))
        existing = filed.get((job, classification))
        if existing is None:
            return {"found": False, "issue_number": None, "url": None, "error": None}
        return {"found": True, "issue_number": existing, "url": f"https://scm/issues/{existing}", "error": None}

    def fake_create(cfg, context):  # type: ignore[no-untyped-def]
        calls["creates"].append(context)
        number = 40 + len(calls["creates"])
        filed[(context["job"], context["classification"])] = number
        return {"issue_number": number, "url": f"https://scm/issues/{number}", "error": None}

    monkeypatch.setattr(module.hermes_incident_issue, "find_open_incident_issue", fake_find)
    monkeypatch.setattr(module.hermes_incident_issue, "create_incident_issue", fake_create)


def _issue_env(monkeypatch, cfg=None, **kwargs):  # type: ignore[no-untyped-def]
    calls: dict = {"lookups": [], "creates": []}
    env = _prepare(monkeypatch, cfg=cfg if cfg is not None else _issue_settings(), **kwargs)
    _install_issue_tracker(monkeypatch, calls)
    env.calls.update(calls)
    return env


def _human_required_output(build_number: int = 12) -> str:
    return _model_output(
        incident_id=f"{JOB}/{build_number}",
        classification="unknown_build_failure",
        requested_action=None,
        human_required=True,
        reason="the failure matches no known signature",
    )


def _issue_events(storage):  # type: ignore[no-untyped-def]
    return _events(storage, module.hermes_incident_issue.ISSUE_EVENT_TYPE)


def test_human_required_incident_files_one_issue(monkeypatch) -> None:
    env = _issue_env(monkeypatch, run=_run(output=_human_required_output()))

    summary = module.run_hermes_autotriage(env.storage)

    assert summary["jobs"][JOB]["status"] == "human_required"
    assert env.calls["lookups"] == [("bstein", JOB, JOB, "unknown_build_failure", INCIDENT_ID)]
    context = env.calls["creates"][0]
    assert context["incident_id"] == INCIDENT_ID
    assert context["classification"] == "unknown_build_failure"
    assert context["reason"] == "the failure matches no known signature"
    assert context["run_id"] == "run-1"
    assert _issue_events(env.storage) == [
        {
            "incident_id": INCIDENT_ID,
            "job": JOB,
            "build_number": 12,
            "classification": "unknown_build_failure",
            "issue_number": 41,
            "url": "https://scm/issues/41",
            "skipped": False,
            "error": None,
        }
    ]
    assert _statuses(env.storage)[-1] == "human_required"


def test_second_incident_with_the_same_classification_skips(monkeypatch) -> None:
    env = _issue_env(monkeypatch, run=_run(output=_human_required_output()))
    module.run_hermes_autotriage(env.storage)

    _prepare(
        monkeypatch,
        cfg=_issue_settings(),
        last_build=_build(13, "FAILURE"),
        run=_run(output=_human_required_output(13)),
        storage=env.storage,
    )
    module.run_hermes_autotriage(env.storage)

    assert len(env.calls["creates"]) == 1
    events = _issue_events(env.storage)
    assert [event["skipped"] for event in events] == [False, True]
    assert events[1]["incident_id"] == f"{JOB}/13"
    assert events[1]["issue_number"] == 41


def test_unmapped_job_and_disabled_switch_file_nothing(monkeypatch) -> None:
    for cfg in (_issue_settings(hermes_issue_repos={}), _issue_settings(hermes_issues_enabled=False)):
        env = _issue_env(monkeypatch, cfg=cfg, run=_run(output=_human_required_output()))

        module.run_hermes_autotriage(env.storage)

        assert env.calls["lookups"] == []
        assert env.calls["creates"] == []
        assert _issue_events(env.storage) == []


def test_remediated_incident_files_no_issue(monkeypatch) -> None:
    env = _issue_env(monkeypatch)

    summary = module.run_hermes_autotriage(env.storage)

    assert summary["jobs"][JOB]["status"] == "awaiting_rebuild"
    assert env.calls["creates"] == []
    assert _issue_events(env.storage) == []


def test_issue_filing_failure_never_breaks_the_tick(monkeypatch) -> None:
    env = _issue_env(monkeypatch, run=_run(output=_human_required_output()))

    def explode(*args, **kwargs):  # type: ignore[no-untyped-def]
        raise RuntimeError("gitea is unreachable")

    monkeypatch.setattr(module.hermes_incident_issue, "maybe_file_issue", explode)

    summary = module.run_hermes_autotriage(env.storage)

    assert summary["status"] == "ok"
    assert summary["jobs"][JOB]["status"] == "human_required"
    assert _statuses(env.storage) == ["detected", "diagnosed", "human_required"]