ariadne/tests/test_hermes_autotriage.py
codex 281f06dccd
All checks were successful
Tests / Declarative: Post Actions passed: 1192
feat(hermes): triage the branches inside multibranch folders
A multibranch project is a folder, not a job: its /api/json carries no
lastBuild at all, only a jobs array with one child per branch. Detection read
lastBuild, so hermes-code-demo-branches was returned as skipped on every tick
since it was created and no branch could ever be triaged.

Expand a folder into its branch jobs and process each. Jenkins addresses them
as parent/job/branch, which the existing fetch already builds correctly, and
branch names stay URL-encoded because re-encoding them yields a 404.

Expansion is capped by ARIADNE_HERMES_MAX_BRANCHES (default 5): a repository
with many active branches would otherwise multiply the watched job count
without limit and could open one incident per failing branch in a single tick.
A name containing a slash is refused outright, since it would escape the
parent and address an unrelated job.

The Jenkins transport moves to its own module so the orchestrator reads as
decision logic, and jobs[name] rides along on the existing request rather than
costing a second call per job per tick.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-08-06 11:01:57 -03:00

483 lines
20 KiB
Python

from __future__ import annotations
import time
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 "The Jenkins job under analysis is hermes-triage-demo." in prompt
assert "reserved for the job hermes-triage-demo" 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_within_the_cap_is_skipped(monkeypatch) -> None:
"""A build that is merely still running is not triage's business."""
fresh = int(time.time() * 1000)
env = _prepare(monkeypatch, last_build=_build(12, None, building=True, timestamp=fresh))
assert module.run_hermes_autotriage(env.storage)["jobs"][JOB] == {"status": "skipped"}
assert env.storage.events == []
def test_building_build_past_the_cap_escalates(monkeypatch) -> None:
"""A build that never terminates must not stay invisible.
It files no incident, raises no alert, and holds a Jenkins agent slot the
whole time, so the only signal is that it has run too long.
"""
env = _prepare(monkeypatch, last_build=_build(12, None, building=True))
summary = module.run_hermes_autotriage(env.storage)["jobs"][JOB]
assert summary == {
"status": "human_required",
"incident_id": INCIDENT_ID,
"reason": module.hermes_hung_builds.HUNG_REASON,
}
assert env.storage.events != []
def test_hung_build_is_not_escalated_twice(monkeypatch) -> None:
"""Every tick sees the same running build; only the first may escalate."""
env = _prepare(monkeypatch, last_build=_build(12, None, building=True))
module.run_hermes_autotriage(env.storage)
before = len(env.storage.events)
second = module.run_hermes_autotriage(env.storage)["jobs"][JOB]
assert second == {"status": "deduped", "incident_id": INCIDENT_ID}
assert len(env.storage.events) == before
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"
tree = params["tree"]
assert "lastBuild[number,result,building,timestamp,duration,url]" in tree
# jobs[name] rides along on the same request so a multibranch folder, which
# carries no lastBuild, is recognisable without a second call per tick.
assert "jobs[name]" in tree
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"]