bstein/ariadne
1
0
Fork 0
Code Issues 3 Pull Requests 1 Actions Packages Projects Releases Wiki Activity

feat(hermes-triage): retry_transient_infra action for connectivity failures

Browse Source 
Second entry in the action registry, proving it is a real extension point.
No cluster mutation: the action is one Jenkins rebuild.

- hermes_infra_signals: reviewable marker set across DNS/connectivity,
  image pull, upstream 5xx and agent-channel loss; Ariadne independently
  confirms a marker in the evidence before any retry, and records which
  marker justified it. "no space left on device" is deliberately excluded
  because a retry lands on the same full volume.
- decision: classification -> action registry (action_classifications),
  falling back to the previous single-classification behavior
- repair: retry_build posts to /build for unparameterized real jobs and
  buildWithParameters for the fixture demo job
- orchestrator: retry path records requested/accepted/executed and moves
  the incident to awaiting_rebuild so the existing success path resolves
  it; one action per incident still enforced, so a retry cannot loop
- events layer split out of the orchestrator to stay under the LOC cap

Motivated by real failures tonight: pip DNS resolution and a Gitea 443
connect timeout, plus live incident metis/271 (SCM checkout timeout).

30 new tests; 368 pass in the hermes suite.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
codex 2026-08-05 20:42:48 -03:00
parent d401cf56a2
commit 8c65b7bd60
14 changed files with 1322 additions and 386 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

281
ariadne/services/hermes_autotriage.py
View File

@ -8,34 +8,34 @@ import httpx
from ..settings import settings
from ..utils.logging import get_logger
from . import hermes_agent_client, hermes_autotriage_repair, hermes_code_flow
from . import hermes_agent_client, hermes_autotriage_repair, hermes_code_flow, 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,
INCIDENT_STATUSES,
refresh_incident_gauges,
set_incident_gauge,
)
logger = get_logger(__name__)
INCIDENT_EVENT_TYPE = "hermes_autotriage_incident"
DIAGNOSIS_EVENT_TYPE = "hermes_autotriage_diagnosis"
ACTION_EVENT_TYPE = "hermes_autotriage_action"
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"
KNOWN_ACTION_IDS = (REPAIR_ACTION, RETRY_ACTION)
REBUILD_FAILED_REASON = "repair rebuild failed"
CODE_FIX_PROPOSED_REASON = "code_fix_proposed"
_LAST_BUILD_TREE = "lastBuild[number,result,building,timestamp,duration,url]"
_EVENT_SCAN_LIMIT = 500
_RUN_COMPLETED = "completed"
_UNKNOWN_ACTION_LABEL = "unknown"
_ACTION_COUNTER_RESULTS = {"executed": "success"}
_PROMPT_TEMPLATE = """Use $triage-titan-test-failures.
Analyze incident __INCIDENT_ID__.
@ -48,6 +48,8 @@ Return ONLY a single JSON object with exactly these keys and no others:
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:
@ -64,7 +66,7 @@ def run_hermes_autotriage(storage: Any) -> dict[str, Any]:
if not settings.hermes_autotriage_enabled:
return {"status": "disabled"}
started = time.time()
incidents = _incident_state(storage)
incidents = hermes_events.incident_state(storage)
jobs: dict[str, Any] = {}
for job in settings.hermes_autotriage_job_allowlist:
jobs[job] = _process_job(storage, job, incidents)
@ -76,34 +78,6 @@ def run_hermes_autotriage(storage: Any) -> dict[str, Any]:
return {"status": "ok", "jobs": jobs}
def _incident_state(storage: Any) -> dict[str, dict[str, Any]]:
"""Fold incident events into the latest state per incident id."""
rows = storage.list_events(limit=_EVENT_SCAN_LIMIT, event_type=INCIDENT_EVENT_TYPE)
incidents: dict[str, dict[str, Any]] = {}
for row in rows:
detail = _event_detail(row)
incident_id = str(detail.get("incident_id") or "") if detail else ""
if incident_id and incident_id not in incidents:
incidents[incident_id] = detail or {}
return incidents
def _event_detail(row: Any) -> dict[str, Any] | None:
"""Return an event row's detail as a dict, decoding stored JSON."""
detail = row.get("detail") if isinstance(row, dict) else None
if isinstance(detail, dict):
return detail
if isinstance(detail, str):
try:
payload = json.loads(detail)
except json.JSONDecodeError:
return None
return payload if isinstance(payload, dict) else None
return None
def _process_job(storage: Any, job: str, incidents: dict[str, dict[str, Any]]) -> dict[str, Any]:
"""Inspect one allowlisted job's last build and advance its incidents."""
@ -167,8 +141,8 @@ def _resolve_on_success(
and incident.get("status") == "awaiting_rebuild"
and _int_value(incident.get("build_number")) < number
):
base = _incident_base(incident)
_record_incident(storage, base, "resolved", {"resolved_by_build": 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}
@ -211,14 +185,14 @@ def _mark_rebuild_failure(
) -> dict[str, Any]:
"""Fail the incident whose rebuild broke and escalate the new failure."""
stale_base = _incident_base(stale)
_record_incident(
stale_base = hermes_events.incident_base(stale)
hermes_events.record_incident(
storage,
stale_base,
"failed",
{"reason": REBUILD_FAILED_REASON, "failed_rebuild": base["incident_id"]},
)
_record_incident(storage, base, "human_required", {"reason": REBUILD_FAILED_REASON})
hermes_events.record_incident(storage, base, "human_required", {"reason": REBUILD_FAILED_REASON})
return {
"status": "rebuild_failed",
"incident_id": str(base["incident_id"]),
@ -232,7 +206,7 @@ def _run_pipeline(
"""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"))}
_record_incident(
hermes_events.record_incident(
storage,
base,
"detected",
@ -248,26 +222,53 @@ def _run_pipeline(
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}"
_record_diagnosis(storage, base, run, None, (False, reason))
_record_incident(storage, base, "human_required", {"reason": reason})
hermes_events.record_diagnosis(storage, base, run, None, hermes_events.Authorization(False, reason))
hermes_events.record_incident(storage, base, "human_required", {"reason": reason})
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)
def _authorize_and_execute(
storage: Any, base: dict[str, Any], run: Any, outcome: Any, bundle: dict[str, Any]
) -> dict[str, Any]:
"""Gate the parsed diagnosis and run its action when every gate passes."""
incident_id = str(base["incident_id"])
matched, marker = _evidence_signature(outcome, bundle, incident_id)
allowed, reason = hermes_decision.authorize_action(
outcome,
_decision_config(),
_prior_action_count(storage, incident_id),
hermes_events.prior_action_count(storage, incident_id),
build_is_terminal_failure=True,
job_allowlisted=True,
evidence_has_signature=hermes_evidence.evidence_has_signature(bundle, incident_id),
evidence_has_signature=matched,
)
hermes_events.record_diagnosis(
storage, base, run, outcome, hermes_events.Authorization(allowed, reason, marker)
)
_record_diagnosis(storage, base, run, outcome, (allowed, reason))
if outcome.valid:
_record_incident(storage, base, "diagnosed", _outcome_phase(outcome))
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()
_record_incident(storage, base, "human_required", {"reason": reason})
hermes_events.record_incident(storage, base, "human_required", {"reason": reason})
return {"status": "human_required", "incident_id": incident_id, "reason": reason}
return _execute_action(storage, base, outcome)
return _execute_action(storage, base, outcome, marker)
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]:
@ -298,17 +299,27 @@ def _propose_code_fix(storage: Any, base: dict[str, Any], bundle: dict[str, Any]
else:
reason = str(result.get("reason") or "code_fix_not_proposed")
phase = {"reason": reason}
_record_incident(storage, base, "human_required", phase)
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) -> dict[str, Any]:
"""Run the authorized repair action and request the verification rebuild."""
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)
_record_action(storage, base, action_id, "requested", None)
_record_action(storage, base, action_id, "accepted", None)
_record_incident(storage, base, "repairing", {"action": action_id})
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"])
@ -319,8 +330,8 @@ def _execute_action(storage: Any, base: dict[str, Any], outcome: Any) -> dict[st
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"))
_record_action(storage, base, action_id, "executed", {"repair_job": repair.get("job_name")})
_record_incident(
hermes_events.record_action(storage, base, action_id, "executed", {"repair_job": repair.get("job_name")})
hermes_events.record_incident(
storage,
base,
"awaiting_rebuild",
@ -333,11 +344,41 @@ def _execute_action(storage: Any, base: dict[str, Any], outcome: Any) -> dict[st
}
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."""
_record_action(storage, base, action_id, "failed", {"error": error})
_record_incident(storage, base, "failed", {"reason": error}, extra_statuses=("human_required",))
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}
@ -348,112 +389,31 @@ def _build_prompt(incident_id: str, bundle: dict[str, Any]) -> str:
return _PROMPT_TEMPLATE.replace("__INCIDENT_ID__", incident_id).replace("__BUNDLE__", compact)
def _record_incident(
storage: Any,
base: dict[str, Any],
status: str,
phase: dict[str, Any] | None = None,
extra_statuses: tuple[str, ...] = (),
) -> None:
"""Append an incident event and publish its one-hot status gauge."""
storage.record_event(INCIDENT_EVENT_TYPE, {**base, "status": status, "phase": phase or {}})
set_incident_gauge(str(base["job"]), str(base["build_number"]), {status, *extra_statuses})
def _record_action(
storage: Any,
base: dict[str, Any],
action_id: str,
result: str,
detail: dict[str, Any] | None,
) -> None:
"""Append an action event and increment the bounded action counter."""
HERMES_TRIAGE_ACTION_TOTAL.labels(
action=action_id, result=_ACTION_COUNTER_RESULTS.get(result, result)
).inc()
payload: dict[str, Any] = {**base, "action": action_id, "result": result}
if detail:
payload["detail"] = detail
storage.record_event(ACTION_EVENT_TYPE, payload)
def _record_diagnosis(
storage: Any,
base: dict[str, Any],
run: Any,
outcome: Any,
authorization: tuple[bool, str],
) -> None:
"""Append a diagnosis event with run metadata and the parsed outcome."""
allowed, reason = authorization
storage.record_event(
DIAGNOSIS_EVENT_TYPE,
{
**base,
"run": {
"status": run.status,
"run_id": run.run_id,
"session_id": run.session_id,
"error": run.error,
"duration_seconds": run.duration_seconds,
"denied_approvals": run.denied_approvals,
},
"outcome": _outcome_phase(outcome) if outcome is not None else None,
"authorized": allowed,
"authorize_reason": reason,
},
)
def _outcome_phase(outcome: Any) -> dict[str, Any]:
"""Summarize a DecisionOutcome for event details."""
decision = outcome.decision
if decision is None:
return {"valid": False, "reject_reason": outcome.reject_reason}
return {
"valid": outcome.valid,
"classification": decision.classification,
"confidence": decision.confidence,
"first_failed_gate": decision.first_failed_gate,
"human_required": decision.human_required,
"requested_action": None if decision.requested_action is None else decision.requested_action.id,
}
def _action_label(outcome: Any) -> str:
"""Return a bounded metric label for the requested action id."""
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
if action is not None and action.id in settings.hermes_allowed_actions:
return action.id
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_IDS):
return action_id
return _UNKNOWN_ACTION_LABEL
def _prior_action_count(storage: Any, incident_id: str) -> int:
"""Count previously recorded action events for one incident."""
def _is_parameterized_job(job: str) -> bool:
"""Report whether a Jenkins job declares build parameters."""
rows = storage.list_events(limit=_EVENT_SCAN_LIMIT, event_type=ACTION_EVENT_TYPE)
count = 0
for row in rows:
detail = _event_detail(row)
if detail is not None and detail.get("incident_id") == incident_id:
count += 1
return count
def _incident_base(incident: dict[str, Any]) -> dict[str, Any]:
"""Normalize a stored incident detail into the base identity fields."""
return {
"incident_id": str(incident.get("incident_id") or ""),
"job": str(incident.get("job") or ""),
"build_number": _int_value(incident.get("build_number")),
}
return job in list(settings.hermes_parameterized_jobs)
def _hermes_run_config() -> dict[str, Any]:
@ -474,6 +434,7 @@ def _decision_config() -> dict[str, Any]:
"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,
}

47
ariadne/services/hermes_autotriage_decision.py
View File

@ -28,6 +28,7 @@ _CONFIDENCE_MAX = 1.0
_DEFAULT_MIN_CONFIDENCE = 1.0
_DEFAULT_MAX_ACTIONS = 1
_DEFAULT_EXPECTED_CLASSIFICATION = "known_demo_fixture_failure"
_ACTION_CLASSIFICATIONS_KEY = "action_classifications"
@dataclass(frozen=True)
@ -129,7 +130,11 @@ def authorize_action( # noqa: PLR0913 - gate signature is part of the frozen mo
Inputs: the parsed DecisionOutcome, gate config, and deterministic facts
established outside the model (prior action count, build terminality,
job allowlist membership, evidence signature presence).
job allowlist membership, evidence signature presence). `cfg` may carry
`action_classifications`, a classification -> action id registry; when it
is present and non-empty the classification must be a registered key and
the requested action must be exactly the action mapped to it (and still
allowlisted). Without it the single `expected_classification` is enforced.
Outputs: (allowed, reason); reason is "authorized" only when every gate
passes, otherwise it names the first failing gate.
"""
@ -138,7 +143,6 @@ def authorize_action( # noqa: PLR0913 - gate signature is part of the frozen mo
return False, f"response_invalid: {outcome.reject_reason or 'missing_decision'}"
decision = outcome.decision
action = decision.requested_action
expected_classification = str(cfg.get("expected_classification") or _DEFAULT_EXPECTED_CLASSIFICATION)
allowed_actions = [str(item) for item in (cfg.get("allowed_actions") or [])]
min_confidence = _float_value(cfg.get("min_confidence"), _DEFAULT_MIN_CONFIDENCE)
max_actions = _int_value(cfg.get("max_actions_per_incident"), _DEFAULT_MAX_ACTIONS)
@ -146,13 +150,11 @@ def authorize_action( # noqa: PLR0913 - gate signature is part of the frozen mo
(not outcome.human_required and not decision.human_required, "human_required"),
(build_is_terminal_failure, "build_not_terminal_failure"),
(job_allowlisted, "job_not_allowlisted"),
(
decision.classification == expected_classification,
f"classification_mismatch: got {decision.classification!r} expected {expected_classification!r}",
),
_classification_gate(decision, cfg),
(action is not None, "requested_action_missing"),
(action is None or action.type == RUN_ARIADNE_JOB_ACTION, "requested_action_type_invalid"),
(action is None or action.id in allowed_actions, f"action_not_allowlisted: {_action_id(action)!r}"),
_action_classification_gate(decision, action, cfg),
(
decision.confidence >= min_confidence,
f"confidence_below_minimum: {decision.confidence} < {min_confidence}",
@ -167,6 +169,39 @@ def authorize_action( # noqa: PLR0913 - gate signature is part of the frozen mo
return True, AUTHORIZED_REASON
def _action_classifications(cfg: dict) -> dict[str, str]:
raw = cfg.get(_ACTION_CLASSIFICATIONS_KEY)
if not isinstance(raw, dict):
return {}
return {str(key): str(value) for key, value in raw.items() if str(key) and str(value)}
def _classification_gate(decision: TriageDecision, cfg: dict) -> tuple[bool, str]:
mapping = _action_classifications(cfg)
if not mapping:
expected = str(cfg.get("expected_classification") or _DEFAULT_EXPECTED_CLASSIFICATION)
return (
decision.classification == expected,
f"classification_mismatch: got {decision.classification!r} expected {expected!r}",
)
return (
decision.classification in mapping,
f"classification_not_supported: {decision.classification!r}",
)
def _action_classification_gate(
decision: TriageDecision, action: RequestedAction | None, cfg: dict
) -> tuple[bool, str]:
expected_action = _action_classifications(cfg).get(decision.classification)
if expected_action is None or action is None:
return True, ""
return (
action.id == expected_action,
f"action_does_not_match_classification: {action.id!r} expected {expected_action!r}",
)
def _rejected(reason: str) -> DecisionOutcome:
return DecisionOutcome(valid=False, decision=None, human_required=True, reject_reason=reason)

210
ariadne/services/hermes_autotriage_events.py Normal file
View File

@ -0,0 +1,210 @@
"""Event-log reads and writes for the Hermes auto-triage incident timeline.
Incidents, diagnoses, and actions are appended to the shared Ariadne event
log; folding those rows back into the latest state per incident is what makes
the scheduler tick idempotent and what enforces the one-action-per-incident
budget across pod restarts.
"""
from __future__ import annotations
from dataclasses import dataclass
import json
from typing import Any
from .hermes_autotriage_metrics import HERMES_TRIAGE_ACTION_TOTAL, set_incident_gauge
INCIDENT_EVENT_TYPE = "hermes_autotriage_incident"
DIAGNOSIS_EVENT_TYPE = "hermes_autotriage_diagnosis"
ACTION_EVENT_TYPE = "hermes_autotriage_action"
_EVENT_SCAN_LIMIT = 500
_ACTION_COUNTER_RESULTS = {"executed": "success"}
@dataclass(frozen=True)
class Authorization:
"""Represent the authorization verdict recorded on a diagnosis event.
Inputs: the gate-chain verdict, the first failing gate name (or
"authorized"), and the evidence marker that satisfied the signature gate
when the action's signature check names one. Outputs: the fields written
onto the diagnosis event so the audit trail records why an action was
allowed.
"""
allowed: bool
reason: str
evidence_marker: str | None = None
def incident_state(storage: Any) -> dict[str, dict[str, Any]]:
"""Fold incident events into the latest state per incident id.
Inputs: a storage object providing list_events. Outputs: a map of
incident id to its most recent event detail.
"""
rows = storage.list_events(limit=_EVENT_SCAN_LIMIT, event_type=INCIDENT_EVENT_TYPE)
incidents: dict[str, dict[str, Any]] = {}
for row in rows:
detail = event_detail(row)
incident_id = str(detail.get("incident_id") or "") if detail else ""
if incident_id and incident_id not in incidents:
incidents[incident_id] = detail or {}
return incidents
def event_detail(row: Any) -> dict[str, Any] | None:
"""Return an event row's detail as a dict, decoding stored JSON.
Inputs: one event row. Outputs: the detail dict, or None when the row or
its detail is missing, malformed, or not an object.
"""
detail = row.get("detail") if isinstance(row, dict) else None
if isinstance(detail, dict):
return detail
if isinstance(detail, str):
try:
payload = json.loads(detail)
except json.JSONDecodeError:
return None
return payload if isinstance(payload, dict) else None
return None
def prior_action_count(storage: Any, incident_id: str) -> int:
"""Count previously recorded action events for one incident.
Inputs: a storage object providing list_events and an incident id.
Outputs: the number of action events already recorded, which is what
keeps one incident to a single remediation attempt.
"""
rows = storage.list_events(limit=_EVENT_SCAN_LIMIT, event_type=ACTION_EVENT_TYPE)
count = 0
for row in rows:
detail = event_detail(row)
if detail is not None and detail.get("incident_id") == incident_id:
count += 1
return count
def incident_base(incident: dict[str, Any]) -> dict[str, Any]:
"""Normalize a stored incident detail into the base identity fields.
Inputs: a stored incident event detail. Outputs: {"incident_id", "job",
"build_number"} with coerced types.
"""
return {
"incident_id": str(incident.get("incident_id") or ""),
"job": str(incident.get("job") or ""),
"build_number": _int_value(incident.get("build_number")),
}
def record_incident(
storage: Any,
base: dict[str, Any],
status: str,
phase: dict[str, Any] | None = None,
extra_statuses: tuple[str, ...] = (),
) -> None:
"""Append an incident event and publish its one-hot status gauge.
Inputs: storage, the incident identity fields, the new status, optional
phase detail, and any extra statuses that should also read 1 on the
gauge. Outputs: none.
"""
storage.record_event(INCIDENT_EVENT_TYPE, {**base, "status": status, "phase": phase or {}})
set_incident_gauge(str(base["job"]), str(base["build_number"]), {status, *extra_statuses})
def record_action(
storage: Any,
base: dict[str, Any],
action_id: str,
result: str,
detail: dict[str, Any] | None,
) -> None:
"""Append an action event and increment the bounded action counter.
Inputs: storage, the incident identity fields, the allowlisted action id,
the lifecycle result (requested/accepted/executed/failed), and optional
detail. Outputs: none.
"""
HERMES_TRIAGE_ACTION_TOTAL.labels(
action=action_id, result=_ACTION_COUNTER_RESULTS.get(result, result)
).inc()
payload: dict[str, Any] = {**base, "action": action_id, "result": result}
if detail:
payload["detail"] = detail
storage.record_event(ACTION_EVENT_TYPE, payload)
def record_diagnosis(
storage: Any,
base: dict[str, Any],
run: Any,
outcome: Any,
authorization: Authorization,
) -> None:
"""Append a diagnosis event with run metadata and the parsed outcome.
Inputs: storage, the incident identity fields, the Hermes run result, the
parsed DecisionOutcome (or None when the run never completed), and the
Authorization verdict. Outputs: none.
"""
storage.record_event(
DIAGNOSIS_EVENT_TYPE,
{
**base,
"run": {
"status": run.status,
"run_id": run.run_id,
"session_id": run.session_id,
"error": run.error,
"duration_seconds": run.duration_seconds,
"denied_approvals": run.denied_approvals,
},
"outcome": outcome_phase(outcome) if outcome is not None else None,
"authorized": authorization.allowed,
"authorize_reason": authorization.reason,
"evidence_marker": authorization.evidence_marker,
},
)
def outcome_phase(outcome: Any) -> dict[str, Any]:
"""Summarize a DecisionOutcome for event details.
Inputs: a DecisionOutcome. Outputs: the bounded subset of decision fields
recorded on incident and diagnosis events.
"""
decision = outcome.decision
if decision is None:
return {"valid": False, "reject_reason": outcome.reject_reason}
return {
"valid": outcome.valid,
"classification": decision.classification,
"confidence": decision.confidence,
"first_failed_gate": decision.first_failed_gate,
"human_required": decision.human_required,
"requested_action": None if decision.requested_action is None else decision.requested_action.id,
}
def _int_value(value: Any) -> int:
"""Coerce a value to int, defaulting to zero."""
try:
return int(value)
except (TypeError, ValueError):
return 0

38
ariadne/services/hermes_autotriage_repair.py
View File

@ -19,6 +19,9 @@ _DEFAULT_WAIT_TIMEOUT_SECONDS = 120.0
_POLL_INTERVAL_SECONDS = 2.0
_JOB_TTL_SECONDS = 3600
_REPAIR_MESSAGE = "fixture state reset to healthy"
_PARAMETERIZED_ENDPOINT = "buildWithParameters"
_PLAIN_ENDPOINT = "build"
_BUILD_PARAMETERS = {"SEED_FAILURE": "false"}
def execute_repair(cfg: dict, incident_id: str, build_number: int) -> dict[str, Any]:
@ -57,20 +60,43 @@ def trigger_rebuild(config: Any, job: str) -> dict[str, Any]:
from buildWithParameters counts as success. Never raises.
"""
return _post_build(config, job, _PARAMETERIZED_ENDPOINT, dict(_BUILD_PARAMETERS), "rebuild")
def retry_build(config: Any, job: str, parameterized: bool) -> dict[str, Any]:
"""Re-run a Jenkins job that failed for a transient infrastructure reason.
Inputs: a settings-like object exposing jenkins_base_url,
jenkins_api_user, jenkins_api_token, and jenkins_api_timeout_sec; the
Jenkins job name; and whether that job declares build parameters.
Outputs: {"requested", "error"}; parameterized jobs are retried through
buildWithParameters with the demo failure seed disabled, every other job
through the plain build endpoint because posting parameters to an
unparameterized job is an error. Only an HTTP 201 counts as success.
Never raises.
"""
if parameterized:
return _post_build(config, job, _PARAMETERIZED_ENDPOINT, dict(_BUILD_PARAMETERS), "retry")
return _post_build(config, job, _PLAIN_ENDPOINT, None, "retry")
def _post_build(
config: Any, job: str, endpoint: str, data: dict[str, str] | None, label: str
) -> dict[str, Any]:
"""Post one Jenkins build trigger and map the response to a result dict."""
base_url = str(getattr(config, "jenkins_base_url", "") or "").strip().rstrip("/")
if not base_url:
return {"requested": False, "error": "jenkins base url is empty"}
try:
with httpx.Client(**_jenkins_client_kwargs(config)) as client:
response = client.post(
f"{base_url}/job/{job}/buildWithParameters",
data={"SEED_FAILURE": "false"},
)
response = client.post(f"{base_url}/job/{job}/{endpoint}", data=data)
except Exception as exc:
return {"requested": False, "error": f"rebuild request failed: {exc}"}
return {"requested": False, "error": f"{label} request failed: {exc}"}
if response.status_code == HTTP_CREATED:
return {"requested": True, "error": None}
return {"requested": False, "error": f"rebuild http {response.status_code}"}
return {"requested": False, "error": f"{label} http {response.status_code}"}
def _job_payload(cfg: dict, job_name: str, incident_id: str) -> dict[str, Any]:

104
ariadne/services/hermes_infra_signals.py Normal file
View File

@ -0,0 +1,104 @@
"""Detect transient infrastructure signatures in a Hermes evidence bundle.
A transient infrastructure failure is one where the build broke for a reason
outside the repository under test - DNS, connectivity, TLS, container
registry, or build-agent channel errors - so re-running the same commit is
the correct and sufficient response.
`INFRA_MARKERS` is deliberately reviewable: every entry is a literal console
substring an operator can grep for, and the matched marker is returned so the
audit trail records why a build was considered retryable.
Deliberate exclusion: "no space left on device" is NOT a transient marker.
Disk exhaustion is an infrastructure fault, but it is not safely retryable -
the retry lands on the same full volume, so it either fails identically or
hides a capacity problem a human must fix. Any other signal with that
property (a resource a rebuild cannot restore) stays out of this set too.
"""
from __future__ import annotations
from typing import Any
INFRA_MARKERS: tuple[str, ...] = (
# name resolution and connectivity
"temporary failure in name resolution",
"could not resolve host",
"failed to establish a new connection",
"failed to connect to",
"connection refused",
"connection reset by peer",
"connection timed out",
"read timed out",
"tls handshake timeout",
"i/o timeout",
# container registry / image pull
"errimagepull",
"imagepullbackoff",
"manifest unknown",
# upstream service errors from a registry, SCM, or proxy
"500 internal server error",
"502 bad gateway",
"503 service unavailable",
"504 gateway",
# build agent channel loss
"remote end closed connection",
"channel is already closed",
"agent went offline",
)
def has_transient_infra_signature(bundle: dict) -> tuple[bool, str | None]:
"""Report whether the bundle shows a retryable infrastructure failure.
Inputs: an evidence bundle from `collect_evidence`; the console failure
regions and the console tail are searched case-insensitively.
Outputs: (matched, marker) where marker is the first `INFRA_MARKERS`
entry found, scanning failure regions in chronological order before the
tail, so the earliest enforced failure wins over downstream noise.
Returns (False, None) when nothing matches. Never raises.
"""
try:
for text in _search_texts(bundle):
marker = _first_marker(text)
if marker is not None:
return True, marker
except Exception:
return False, None
return False, None
def _search_texts(bundle: dict) -> list[str]:
"""Return the console texts to scan, earliest failure region first."""
if not isinstance(bundle, dict):
return []
jenkins = bundle.get("jenkins")
if not isinstance(jenkins, dict):
return []
regions = jenkins.get("console_failures")
texts = [
str(region.get("text") or "")
for region in (regions if isinstance(regions, list) else [])
if isinstance(region, dict)
]
texts.append(str(jenkins.get("console_tail") or ""))
return [text for text in texts if text]
def _first_marker(text: str) -> str | None:
"""Return the marker occurring earliest in one console text, if any."""
lowered = text.lower()
best: str | None = None
best_index = -1
for marker in INFRA_MARKERS:
index = lowered.find(marker)
if index < 0:
continue
if best is None or index < best_index:
best = marker
best_index = index
return best

2
ariadne/settings.py
View File

@ -185,6 +185,8 @@ class Settings:
hermes_autotriage_job_allowlist: list[str]
hermes_autoremediation_enabled: bool
hermes_allowed_actions: list[str]
hermes_action_classifications: dict[str, str]
hermes_parameterized_jobs: list[str]
hermes_min_confidence: float
hermes_max_actions_per_incident: int
hermes_api_url: str

20
ariadne/settings_sections.py
View File

@ -254,6 +254,15 @@ def _testing_triage_config() -> dict[str, Any]:
}
def _pair_map(raw: str) -> dict[str, str]:
mapping: dict[str, str] = {}
for item in raw.split(","):
key, _, value = item.partition("=")
if key.strip() and value.strip():
mapping[key.strip()] = value.strip()
return mapping
def _hermes_autotriage_config() -> dict[str, Any]:
return {
"hermes_autotriage_enabled": _env_bool("ARIADNE_HERMES_AUTOTRIAGE_ENABLED", "false"),
@ -268,6 +277,17 @@ def _hermes_autotriage_config() -> dict[str, Any]:
for item in _env("ARIADNE_HERMES_ALLOWED_ACTIONS", "repair_demo_fixture").split(",")
if item.strip()
],
"hermes_action_classifications": _pair_map(
_env(
"ARIADNE_HERMES_ACTION_CLASSIFICATIONS",
"known_demo_fixture_failure=repair_demo_fixture,transient_infra_failure=retry_transient_infra",
)
),
"hermes_parameterized_jobs": [
item.strip()
for item in _env("ARIADNE_HERMES_PARAMETERIZED_JOBS", "hermes-triage-demo").split(",")
if item.strip()
],
"hermes_min_confidence": _env_float("ARIADNE_HERMES_MIN_CONFIDENCE", 0.85),
"hermes_max_actions_per_incident": _env_int("ARIADNE_HERMES_MAX_ACTIONS_PER_INCIDENT", 1),
"hermes_api_url": _env(

259
tests/hermes_autotriage_harness.py Normal file
View File

@ -0,0 +1,259 @@
"""Shared fakes for the Hermes auto-triage orchestrator tests.
Not collected by pytest; imported by test_hermes_autotriage.py and
test_hermes_autotriage_retry.py so both drive the same fake Jenkins, storage,
and remediation executors.
"""
from __future__ import annotations
import json
from types import SimpleNamespace
from ariadne.services import hermes_autotriage as module
from ariadne.services.hermes_agent_client import HermesRunResult
JOB = "hermes-triage-demo"
INCIDENT_ID = f"{JOB}/12"
INFRA_MARKER = "temporary failure in name resolution"
class FakeStorage:
def __init__(self) -> None:
self.events: list[dict] = []
def record_event(self, event_type, detail) -> None: # type: ignore[no-untyped-def]
self.events.append({"event_type": event_type, "detail": detail})
def list_events(self, limit=200, event_type=None): # type: ignore[no-untyped-def]
rows = [
dict(row)
for row in reversed(self.events)
if event_type is None or row["event_type"] == event_type
]
return rows[:limit]
class FakeResponse:
def __init__(self, payload) -> None: # type: ignore[no-untyped-def]
self.payload = payload
def raise_for_status(self) -> None:
return None
def json(self): # type: ignore[no-untyped-def]
return self.payload
def _settings(**overrides) -> SimpleNamespace: # type: ignore[no-untyped-def]
values = {
"hermes_autotriage_enabled": True,
"hermes_autotriage_job_allowlist": [JOB],
"hermes_autoremediation_enabled": True,
"hermes_allowed_actions": ["repair_demo_fixture"],
"hermes_action_classifications": {
"known_demo_fixture_failure": "repair_demo_fixture",
"transient_infra_failure": "retry_transient_infra",
},
"hermes_parameterized_jobs": [JOB],
"hermes_min_confidence": 0.85,
"hermes_max_actions_per_incident": 1,
"hermes_api_url": "http://hermes:8642",
"hermes_api_key": "key",
"hermes_run_timeout_seconds": 420.0,
"hermes_demo_namespace": "hermes-triage-demo",
"hermes_demo_fixture_configmap": "hermes-triage-demo-fixture",
"hermes_repair_image": "busybox:1.37",
"hermes_code_enabled": False,
"hermes_code_job": "hermes-code-demo",
"hermes_code_owner": "bstein",
"hermes_code_repo": "hermes-code-demo",
"hermes_code_base_branch": "master",
"hermes_code_candidate_path": "src/discount.py",
"hermes_code_allowed_prefixes": ["src/"],
"hermes_code_allowed_suffixes": [".py"],
"hermes_code_max_patch_bytes": 4000,
"hermes_code_max_changed_lines": 20,
"hermes_gitea_base_url": "https://scm.example",
"hermes_gitea_token": "gitea-token",
"jenkins_base_url": "https://ci.example",
"jenkins_api_user": "user",
"jenkins_api_token": "token",
"jenkins_api_timeout_sec": 5.0,
}
values.update(overrides)
return SimpleNamespace(**values)
def _build(number: int, result, job: str = JOB, **overrides): # type: ignore[no-untyped-def]
payload = {
"number": number,
"result": result,
"building": False,
"timestamp": 1720000000000,
"duration": 60000,
"url": f"https://ci.example/job/{job}/{number}/",
}
payload.update(overrides)
return payload
def _model_output(**overrides) -> str: # type: ignore[no-untyped-def]
payload = {
"incident_id": INCIDENT_ID,
"classification": "known_demo_fixture_failure",
"confidence": 0.95,
"facts": [
{"statement": "fixture-state-check failed", "source": "jenkins", "reference": "console"}
],
"inferences": [],
"first_failed_gate": "fixture-state-check",
"requested_action": {"type": "run_ariadne_job", "id": "repair_demo_fixture"},
"human_required": False,
"reason": "known fixture failure",
}
payload.update(overrides)
return json.dumps(payload)
def _retry_output(job: str = JOB, build_number: int = 12, **overrides) -> str: # type: ignore[no-untyped-def]
payload = {
"incident_id": f"{job}/{build_number}",
"classification": "transient_infra_failure",
"confidence": 0.95,
"facts": [
{"statement": "pip install hit a DNS failure", "source": "jenkins", "reference": "console"}
],
"inferences": [],
"first_failed_gate": "dependencies",
"requested_action": {"type": "run_ariadne_job", "id": "retry_transient_infra"},
"human_required": False,
"reason": "transient name resolution failure on the agent",
}
payload.update(overrides)
return json.dumps(payload)
def _run(status: str = "completed", output=None, error=None) -> HermesRunResult: # type: ignore[no-untyped-def]
return HermesRunResult(
status=status,
output=output,
run_id="run-1",
session_id="sess-1",
error=error,
duration_seconds=1.5,
denied_approvals=0,
)
def _install_jenkins(monkeypatch, calls, last_build, exc) -> None: # type: ignore[no-untyped-def]
class FakeClient:
def __init__(self, **kwargs) -> None: # type: ignore[no-untyped-def]
calls["client_kwargs"] = kwargs
def __enter__(self): # type: ignore[no-untyped-def]
return self
def __exit__(self, *args) -> None: # type: ignore[no-untyped-def]
return None
def get(self, url, params=None): # type: ignore[no-untyped-def]
calls["gets"].append((url, params))
if exc is not None:
raise exc
return FakeResponse({"lastBuild": last_build})
monkeypatch.setattr(module.httpx, "Client", FakeClient)
def _prepare( # type: ignore[no-untyped-def] # noqa: PLR0913
monkeypatch,
*,
cfg=None,
last_build=None,
jenkins_exc=None,
run=None,
repair=None,
rebuild=None,
retry=None,
signature=True,
infra=None,
storage=None,
):
storage = storage if storage is not None else FakeStorage()
calls: dict = {"gets": [], "triage": [], "repairs": [], "rebuilds": [], "retries": []}
monkeypatch.setattr(module, "settings", cfg if cfg is not None else _settings())
_install_jenkins(
monkeypatch, calls, last_build if last_build is not None else _build(12, "FAILURE"), jenkins_exc
)
monkeypatch.setattr(
module.hermes_evidence,
"collect_evidence",
lambda incident_id, job, build: {
"incident_id": incident_id,
"jenkins": {"job": job},
"log_evidence": {"records": []},
},
)
monkeypatch.setattr(
module.hermes_evidence, "evidence_has_signature", lambda bundle, incident_id: signature
)
monkeypatch.setattr(
module.hermes_infra_signals,
"has_transient_infra_signature",
lambda bundle: infra if infra is not None else (True, INFRA_MARKER),
)
def fake_run_triage(config, prompt): # type: ignore[no-untyped-def]
calls["triage"].append((config, prompt))
return run if run is not None else _run(output=_model_output())
def fake_execute_repair(repair_cfg, incident_id, build_number): # type: ignore[no-untyped-def]
calls["repairs"].append((repair_cfg, incident_id, build_number))
if repair is not None:
return repair
return {"job_name": "hermes-demo-repair-12", "succeeded": True, "error": None}
def fake_trigger_rebuild(rebuild_cfg, job): # type: ignore[no-untyped-def]
calls["rebuilds"].append(job)
return rebuild if rebuild is not None else {"requested": True, "error": None}
def fake_retry_build(retry_cfg, job, parameterized): # type: ignore[no-untyped-def]
calls["retries"].append((job, parameterized))
return retry if retry is not None else {"requested": True, "error": None}
monkeypatch.setattr(module.hermes_agent_client, "run_triage", fake_run_triage)
monkeypatch.setattr(module.hermes_autotriage_repair, "execute_repair", fake_execute_repair)
monkeypatch.setattr(module.hermes_autotriage_repair, "trigger_rebuild", fake_trigger_rebuild)
monkeypatch.setattr(module.hermes_autotriage_repair, "retry_build", fake_retry_build)
return SimpleNamespace(storage=storage, calls=calls)
def _events(storage: FakeStorage, event_type: str) -> list[dict]:
return [row["detail"] for row in storage.events if row["event_type"] == event_type]
def _statuses(storage: FakeStorage) -> list[str]:
return [detail["status"] for detail in _events(storage, module.INCIDENT_EVENT_TYPE)]
def _seed_incident( # type: ignore[no-untyped-def]
storage: FakeStorage, status: str, build_number: int = 12, as_json: bool = False, job: str = JOB
) -> None:
detail = {
"incident_id": f"{job}/{build_number}",
"job": job,
"build_number": build_number,
"status": status,
"phase": {},
}
storage.record_event(module.INCIDENT_EVENT_TYPE, json.dumps(detail) if as_json else detail)
def _gauge(build: str, status: str, job: str = JOB) -> float:
return module.HERMES_TRIAGE_INCIDENT.labels(jenkins_job=job, build=build, status=status)._value.get()
def _counter(action: str, result: str) -> float:
return module.HERMES_TRIAGE_ACTION_TOTAL.labels(action=action, result=result)._value.get()

232
tests/test_hermes_autotriage.py
View File

@ -1,218 +1,22 @@
from __future__ import annotations
import json
from types import SimpleNamespace
import pytest
from ariadne.services import hermes_autotriage as module
from ariadne.services.hermes_agent_client import HermesRunResult
JOB = "hermes-triage-demo"
INCIDENT_ID = f"{JOB}/12"
class FakeStorage:
def __init__(self) -> None:
self.events: list[dict] = []
def record_event(self, event_type, detail) -> None: # type: ignore[no-untyped-def]
self.events.append({"event_type": event_type, "detail": detail})
def list_events(self, limit=200, event_type=None): # type: ignore[no-untyped-def]
rows = [
dict(row)
for row in reversed(self.events)
if event_type is None or row["event_type"] == event_type
]
return rows[:limit]
class FakeResponse:
def __init__(self, payload) -> None: # type: ignore[no-untyped-def]
self.payload = payload
def raise_for_status(self) -> None:
return None
def json(self): # type: ignore[no-untyped-def]
return self.payload
def _settings(**overrides) -> SimpleNamespace: # type: ignore[no-untyped-def]
values = {
"hermes_autotriage_enabled": True,
"hermes_autotriage_job_allowlist": [JOB],
"hermes_autoremediation_enabled": True,
"hermes_allowed_actions": ["repair_demo_fixture"],
"hermes_min_confidence": 0.85,
"hermes_max_actions_per_incident": 1,
"hermes_api_url": "http://hermes:8642",
"hermes_api_key": "key",
"hermes_run_timeout_seconds": 420.0,
"hermes_demo_namespace": "hermes-triage-demo",
"hermes_demo_fixture_configmap": "hermes-triage-demo-fixture",
"hermes_repair_image": "busybox:1.37",
"hermes_code_enabled": False,
"hermes_code_job": "hermes-code-demo",
"hermes_code_owner": "bstein",
"hermes_code_repo": "hermes-code-demo",
"hermes_code_base_branch": "master",
"hermes_code_candidate_path": "src/discount.py",
"hermes_code_allowed_prefixes": ["src/"],
"hermes_code_allowed_suffixes": [".py"],
"hermes_code_max_patch_bytes": 4000,
"hermes_code_max_changed_lines": 20,
"hermes_gitea_base_url": "https://scm.example",
"hermes_gitea_token": "gitea-token",
"jenkins_base_url": "https://ci.example",
"jenkins_api_user": "user",
"jenkins_api_token": "token",
"jenkins_api_timeout_sec": 5.0,
}
values.update(overrides)
return SimpleNamespace(**values)
def _build(number: int, result, **overrides): # type: ignore[no-untyped-def]
payload = {
"number": number,
"result": result,
"building": False,
"timestamp": 1720000000000,
"duration": 60000,
"url": f"https://ci.example/job/{JOB}/{number}/",
}
payload.update(overrides)
return payload
def _model_output(**overrides) -> str: # type: ignore[no-untyped-def]
payload = {
"incident_id": INCIDENT_ID,
"classification": "known_demo_fixture_failure",
"confidence": 0.95,
"facts": [
{"statement": "fixture-state-check failed", "source": "jenkins", "reference": "console"}
],
"inferences": [],
"first_failed_gate": "fixture-state-check",
"requested_action": {"type": "run_ariadne_job", "id": "repair_demo_fixture"},
"human_required": False,
"reason": "known fixture failure",
}
payload.update(overrides)
return json.dumps(payload)
def _run(status: str = "completed", output=None, error=None) -> HermesRunResult: # type: ignore[no-untyped-def]
return HermesRunResult(
status=status,
output=output,
run_id="run-1",
session_id="sess-1",
error=error,
duration_seconds=1.5,
denied_approvals=0,
)
def _install_jenkins(monkeypatch, calls, last_build, exc) -> None: # type: ignore[no-untyped-def]
class FakeClient:
def __init__(self, **kwargs) -> None: # type: ignore[no-untyped-def]
calls["client_kwargs"] = kwargs
def __enter__(self): # type: ignore[no-untyped-def]
return self
def __exit__(self, *args) -> None: # type: ignore[no-untyped-def]
return None
def get(self, url, params=None): # type: ignore[no-untyped-def]
calls["gets"].append((url, params))
if exc is not None:
raise exc
return FakeResponse({"lastBuild": last_build})
monkeypatch.setattr(module.httpx, "Client", FakeClient)
def _prepare( # type: ignore[no-untyped-def]
monkeypatch,
*,
cfg=None,
last_build=None,
jenkins_exc=None,
run=None,
repair=None,
rebuild=None,
signature=True,
):
storage = FakeStorage()
calls: dict = {"gets": [], "triage": [], "repairs": [], "rebuilds": []}
monkeypatch.setattr(module, "settings", cfg if cfg is not None else _settings())
_install_jenkins(
monkeypatch, calls, last_build if last_build is not None else _build(12, "FAILURE"), jenkins_exc
)
monkeypatch.setattr(
module.hermes_evidence,
"collect_evidence",
lambda incident_id, job, build: {
"incident_id": incident_id,
"jenkins": {"job": job},
"log_evidence": {"records": []},
},
)
monkeypatch.setattr(
module.hermes_evidence, "evidence_has_signature", lambda bundle, incident_id: signature
)
def fake_run_triage(config, prompt): # type: ignore[no-untyped-def]
calls["triage"].append((config, prompt))
return run if run is not None else _run(output=_model_output())
def fake_execute_repair(repair_cfg, incident_id, build_number): # type: ignore[no-untyped-def]
calls["repairs"].append((repair_cfg, incident_id, build_number))
if repair is not None:
return repair
return {"job_name": "hermes-demo-repair-12", "succeeded": True, "error": None}
def fake_trigger_rebuild(rebuild_cfg, job): # type: ignore[no-untyped-def]
calls["rebuilds"].append(job)
return rebuild if rebuild is not None else {"requested": True, "error": None}
monkeypatch.setattr(module.hermes_agent_client, "run_triage", fake_run_triage)
monkeypatch.setattr(module.hermes_autotriage_repair, "execute_repair", fake_execute_repair)
monkeypatch.setattr(module.hermes_autotriage_repair, "trigger_rebuild", fake_trigger_rebuild)
return SimpleNamespace(storage=storage, calls=calls)
def _events(storage: FakeStorage, event_type: str) -> list[dict]:
return [row["detail"] for row in storage.events if row["event_type"] == event_type]
def _statuses(storage: FakeStorage) -> list[str]:
return [detail["status"] for detail in _events(storage, module.INCIDENT_EVENT_TYPE)]
def _seed_incident(storage: FakeStorage, status: str, build_number: int = 12, as_json: bool = False) -> None:
detail = {
"incident_id": f"{JOB}/{build_number}",
"job": JOB,
"build_number": build_number,
"status": status,
"phase": {},
}
storage.record_event(module.INCIDENT_EVENT_TYPE, json.dumps(detail) if as_json else detail)
def _gauge(build: str, status: str) -> float:
return module.HERMES_TRIAGE_INCIDENT.labels(jenkins_job=JOB, build=build, status=status)._value.get()
def _counter(action: str, result: str) -> float:
return module.HERMES_TRIAGE_ACTION_TOTAL.labels(action=action, result=result)._value.get()
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:
@ -475,10 +279,10 @@ def test_non_terminal_result_is_ignored(monkeypatch) -> None:
def test_event_detail_tolerates_bad_payloads() -> None:
assert module._event_detail({"detail": "not-json"}) is None
assert module._event_detail({"detail": "[1,2]"}) is None
assert module._event_detail({"detail": 5}) is None
assert module._event_detail("not-a-row") is 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

87
tests/test_hermes_autotriage_decision.py
View File

@ -357,3 +357,90 @@ def test_authorize_falls_back_to_conservative_cfg_values() -> None:
assert allowed is False
assert reason == "confidence_below_minimum: 0.93 < 1.0"
def _mapping_cfg(**overrides) -> dict: # type: ignore[no-untyped-def]
base = {
"allowed_actions": ["repair_demo_fixture", "retry_transient_infra"],
"action_classifications": {
"known_demo_fixture_failure": "repair_demo_fixture",
"transient_infra_failure": "retry_transient_infra",
},
}
base.update(overrides)
return _cfg(**base)
def _mapped_payload(classification: str, action_id: str) -> dict:
return _payload(
classification=classification,
requested_action={"type": "run_ariadne_job", "id": action_id},
)
def test_authorize_accepts_each_registered_classification() -> None:
for classification, action_id in _mapping_cfg()["action_classifications"].items():
outcome = _parse(_mapped_payload(classification, action_id))
assert _authorize(outcome=outcome, cfg=_mapping_cfg()) == (True, "authorized")
def test_authorize_rejects_unregistered_classification() -> None:
outcome = _parse(_mapped_payload("novel_failure", "retry_transient_infra"))
allowed, reason = _authorize(outcome=outcome, cfg=_mapping_cfg())
assert allowed is False
assert reason == "classification_not_supported: 'novel_failure'"
def test_authorize_rejects_action_mapped_to_another_classification() -> None:
outcome = _parse(_mapped_payload("transient_infra_failure", "repair_demo_fixture"))
allowed, reason = _authorize(outcome=outcome, cfg=_mapping_cfg())
assert allowed is False
assert reason == (
"action_does_not_match_classification: 'repair_demo_fixture' expected 'retry_transient_infra'"
)
def test_authorize_still_requires_the_mapped_action_to_be_allowlisted() -> None:
outcome = _parse(_mapped_payload("transient_infra_failure", "retry_transient_infra"))
cfg = _mapping_cfg(allowed_actions=["repair_demo_fixture"])
allowed, reason = _authorize(outcome=outcome, cfg=cfg)
assert allowed is False
assert reason == "action_not_allowlisted: 'retry_transient_infra'"
def test_authorize_reports_missing_action_before_the_mapping_gate() -> None:
outcome = _parse(_payload(classification="transient_infra_failure", requested_action=None))
assert _authorize(outcome=outcome, cfg=_mapping_cfg()) == (False, "requested_action_missing")
def test_authorize_mapping_still_enforces_the_remaining_gates() -> None:
outcome = _parse(_mapped_payload("transient_infra_failure", "retry_transient_infra"))
assert _authorize(outcome=outcome, cfg=_mapping_cfg(), evidence_has_signature=False) == (
False,
"evidence_signature_missing",
)
assert _authorize(outcome=outcome, cfg=_mapping_cfg(), prior_action_count=1) == (
False,
"max_actions_reached",
)
assert _authorize(outcome=outcome, cfg=_mapping_cfg(autoremediation_enabled=False)) == (
False,
"autoremediation_disabled",
)
def test_authorize_ignores_an_empty_or_malformed_mapping() -> None:
for mapping in ({}, None, "known_demo_fixture_failure=repair_demo_fixture", []):
cfg = _cfg(action_classifications=mapping)
assert _authorize(cfg=cfg) == (True, "authorized")
allowed, reason = _authorize(outcome=_parse(_payload(classification="other")), cfg=cfg)
assert allowed is False
assert reason == "classification_mismatch: got 'other' expected 'known_demo_fixture_failure'"

48
tests/test_hermes_autotriage_repair.py
View File

@ -213,3 +213,51 @@ def test_trigger_rebuild_without_credentials(monkeypatch) -> None:
result = module.trigger_rebuild(config, "hermes-triage-demo")
assert result["requested"] is True
assert "auth" not in calls["kwargs"]
def test_retry_build_parameterized_reuses_the_rebuild_form(monkeypatch) -> None:
calls = _install_http(monkeypatch, response=FakeResponse(201))
result = module.retry_build(_jenkins_settings(), "hermes-triage-demo", parameterized=True)
assert result == {"requested": True, "error": None}
url, data = calls["posts"][0]
assert url == "https://ci.example/job/hermes-triage-demo/buildWithParameters"
assert data == {"SEED_FAILURE": "false"}
assert calls["kwargs"]["auth"] == ("user", "token")
def test_retry_build_unparameterized_posts_no_parameters(monkeypatch) -> None:
calls = _install_http(monkeypatch, response=FakeResponse(201))
result = module.retry_build(_jenkins_settings(), "titan-iac", parameterized=False)
assert result == {"requested": True, "error": None}
url, data = calls["posts"][0]
assert url == "https://ci.example/job/titan-iac/build"
assert data is None
def test_retry_build_non_created_status(monkeypatch) -> None:
_install_http(monkeypatch, response=FakeResponse(500))
result = module.retry_build(_jenkins_settings(), "titan-iac", parameterized=False)
assert result == {"requested": False, "error": "retry http 500"}
def test_retry_build_request_failure(monkeypatch) -> None:
_install_http(monkeypatch, exc=RuntimeError("connection reset by peer"))
result = module.retry_build(_jenkins_settings(), "titan-iac", parameterized=False)
assert result == {"requested": False, "error": "retry request failed: connection reset by peer"}
def test_retry_build_without_base_url(monkeypatch) -> None:
calls = _install_http(monkeypatch, response=FakeResponse(201))
result = module.retry_build(_jenkins_settings(jenkins_base_url=""), "titan-iac", parameterized=True)
assert result == {"requested": False, "error": "jenkins base url is empty"}
assert calls["posts"] == []
def test_retry_build_without_credentials(monkeypatch) -> None:
calls = _install_http(monkeypatch, response=FakeResponse(201))
config = _jenkins_settings(jenkins_api_user="", jenkins_api_token="")
result = module.retry_build(config, "titan-iac", parameterized=False)
assert result["requested"] is True
assert "auth" not in calls["kwargs"]

213
tests/test_hermes_autotriage_retry.py Normal file
View File

@ -0,0 +1,213 @@
from __future__ import annotations
from ariadne.services import hermes_autotriage as module
from tests.hermes_autotriage_harness import (
INCIDENT_ID,
INFRA_MARKER,
JOB,
_build,
_counter,
_events,
_gauge,
_prepare,
_retry_output,
_run,
_seed_incident,
_settings,
_statuses,
)
OTHER_JOB = "titan-iac"
BOTH_ACTIONS = ["repair_demo_fixture", "retry_transient_infra"]
def _retry_settings(**overrides): # type: ignore[no-untyped-def]
values = {"hermes_allowed_actions": BOTH_ACTIONS}
values.update(overrides)
return _settings(**values)
def _retry_env(monkeypatch, **overrides): # type: ignore[no-untyped-def]
kwargs = {"cfg": _retry_settings(), "run": _run(output=_retry_output())}
kwargs.update(overrides)
return _prepare(monkeypatch, **kwargs)
def test_transient_infra_failure_is_retried(monkeypatch) -> None:
success_before = _counter("retry_transient_infra", "success")
env = _retry_env(monkeypatch)
summary = module.run_hermes_autotriage(env.storage)
assert summary["jobs"][JOB] == {
"status": "awaiting_rebuild",
"incident_id": INCIDENT_ID,
"action": "retry_transient_infra",
"evidence_marker": INFRA_MARKER,
}
assert _statuses(env.storage) == ["detected", "diagnosed", "awaiting_rebuild"]
assert env.calls["retries"] == [(JOB, True)]
assert env.calls["repairs"] == []
assert env.calls["rebuilds"] == []
assert _counter("retry_transient_infra", "success") == success_before + 1.0
assert _gauge("12", "awaiting_rebuild") == 1.0
def test_retry_records_the_matched_marker_on_every_event(monkeypatch) -> None:
env = _retry_env(monkeypatch)
module.run_hermes_autotriage(env.storage)
actions = _events(env.storage, module.ACTION_EVENT_TYPE)
assert [action["result"] for action in actions] == ["requested", "accepted", "executed"]
assert all(action["action"] == "retry_transient_infra" for action in actions)
assert actions[0]["detail"] == {"evidence_marker": INFRA_MARKER}
assert actions[2]["detail"] == {"evidence_marker": INFRA_MARKER}
diagnosis = _events(env.storage, module.DIAGNOSIS_EVENT_TYPE)[0]
assert diagnosis["authorized"] is True
assert diagnosis["authorize_reason"] == "authorized"
assert diagnosis["evidence_marker"] == INFRA_MARKER
assert diagnosis["outcome"]["classification"] == "transient_infra_failure"
incident = _events(env.storage, module.INCIDENT_EVENT_TYPE)[-1]
assert incident["phase"] == {"action": "retry_transient_infra", "evidence_marker": INFRA_MARKER}
def test_retry_resolves_on_the_next_green_build(monkeypatch) -> None:
env = _retry_env(monkeypatch)
module.run_hermes_autotriage(env.storage)
green = _prepare(
monkeypatch,
cfg=_retry_settings(),
last_build=_build(13, "SUCCESS"),
storage=env.storage,
)
summary = module.run_hermes_autotriage(green.storage)
assert summary["jobs"][JOB] == {"status": "healthy", "resolved": [INCIDENT_ID]}
assert _statuses(env.storage)[-1] == "resolved"
assert _gauge("12", "resolved") == 1.0
def test_unparameterized_job_is_retried_without_parameters(monkeypatch) -> None:
cfg = _retry_settings(
hermes_autotriage_job_allowlist=[OTHER_JOB],
hermes_parameterized_jobs=[JOB],
)
env = _prepare(
monkeypatch,
cfg=cfg,
last_build=_build(12, "FAILURE", job=OTHER_JOB),
run=_run(output=_retry_output(job=OTHER_JOB)),
)
summary = module.run_hermes_autotriage(env.storage)
assert summary["jobs"][OTHER_JOB]["status"] == "awaiting_rebuild"
assert env.calls["retries"] == [(OTHER_JOB, False)]
def test_retry_without_infra_signature_requires_a_human(monkeypatch) -> None:
env = _retry_env(monkeypatch, infra=(False, None))
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["retries"] == []
assert _events(env.storage, module.DIAGNOSIS_EVENT_TYPE)[0]["evidence_marker"] is None
def test_retry_ignores_the_fixture_signature(monkeypatch) -> None:
env = _retry_env(monkeypatch, signature=False)
summary = module.run_hermes_autotriage(env.storage)
assert summary["jobs"][JOB]["status"] == "awaiting_rebuild"
assert env.calls["retries"] == [(JOB, True)]
def test_fixture_action_ignores_the_infra_signature(monkeypatch) -> None:
env = _prepare(monkeypatch, cfg=_retry_settings(), signature=False, infra=(True, INFRA_MARKER))
summary = module.run_hermes_autotriage(env.storage)
assert summary["jobs"][JOB]["reason"] == "evidence_signature_missing"
assert env.calls["retries"] == []
def test_retry_is_capped_at_one_action_per_incident(monkeypatch) -> None:
env = _retry_env(monkeypatch)
_seed_incident(env.storage, "detected")
env.storage.record_event(
module.ACTION_EVENT_TYPE,
{"incident_id": INCIDENT_ID, "action": "retry_transient_infra", "result": "requested"},
)
summary = module.run_hermes_autotriage(env.storage)
assert summary["jobs"][JOB]["reason"] == "max_actions_reached"
assert env.calls["retries"] == []
def test_retry_trigger_failure_marks_failed_and_human(monkeypatch) -> None:
failed_before = _counter("retry_transient_infra", "failed")
env = _retry_env(monkeypatch, retry={"requested": False, "error": "retry http 500"})
summary = module.run_hermes_autotriage(env.storage)
assert summary["jobs"][JOB] == {
"status": "failed",
"incident_id": INCIDENT_ID,
"reason": "retry http 500",
}
assert _statuses(env.storage) == ["detected", "diagnosed", "failed"]
actions = _events(env.storage, module.ACTION_EVENT_TYPE)
assert [action["result"] for action in actions] == ["requested", "accepted", "failed"]
assert _counter("retry_transient_infra", "failed") == failed_before + 1.0
assert _gauge("12", "human_required") == 1.0
def test_retry_action_is_rejected_while_not_allowlisted(monkeypatch) -> None:
rejected_before = _counter("retry_transient_infra", "rejected")
env = _prepare(monkeypatch, run=_run(output=_retry_output()))
summary = module.run_hermes_autotriage(env.storage)
assert summary["jobs"][JOB]["reason"] == "action_not_allowlisted: 'retry_transient_infra'"
assert env.calls["retries"] == []
assert _counter("retry_transient_infra", "rejected") == rejected_before + 1.0
def test_action_must_match_its_classification(monkeypatch) -> None:
output = _retry_output(requested_action={"type": "run_ariadne_job", "id": "repair_demo_fixture"})
env = _retry_env(monkeypatch, run=_run(output=output))
summary = module.run_hermes_autotriage(env.storage)
assert summary["jobs"][JOB]["reason"] == (
"action_does_not_match_classification: 'repair_demo_fixture' expected 'retry_transient_infra'"
)
assert env.calls["retries"] == []
assert env.calls["repairs"] == []
def test_unregistered_classification_requires_a_human(monkeypatch) -> None:
env = _retry_env(monkeypatch, run=_run(output=_retry_output(classification="flaky_network_guess")))
summary = module.run_hermes_autotriage(env.storage)
assert summary["jobs"][JOB]["reason"] == "classification_not_supported: 'flaky_network_guess'"
assert env.calls["retries"] == []
def test_prompt_documents_the_retry_action(monkeypatch) -> None:
env = _retry_env(monkeypatch)
module.run_hermes_autotriage(env.storage)
_, prompt = env.calls["triage"][0]
assert "Use classification transient_infra_failure with requested_action" in prompt
assert '{"type": "run_ariadne_job", "id": "retry_transient_infra"}' in prompt
assert "unrelated to the repository's code or tests" in prompt
assert "Otherwise leave requested_action null" in prompt

132
tests/test_hermes_infra_signals.py Normal file
View File

@ -0,0 +1,132 @@
from __future__ import annotations
import pytest
from ariadne.services import hermes_infra_signals as module
def _bundle(regions=None, tail=None, **jenkins): # type: ignore[no-untyped-def]
payload = {
"job": "titan-iac",
"console_failures": [{"marker": "ERROR:", "line_number": 10, "text": text} for text in regions or []],
"console_tail": tail,
}
payload.update(jenkins)
return {"incident_id": "titan-iac/12", "jenkins": payload, "log_evidence": {"records": []}}
@pytest.mark.parametrize("marker", module.INFRA_MARKERS)
def test_every_marker_is_detected_in_a_region(marker) -> None:
bundle = _bundle(regions=[f"+ pip install -r requirements.txt\n{marker}\nbuild step failed"])
assert module.has_transient_infra_signature(bundle) == (True, marker)
@pytest.mark.parametrize("marker", module.INFRA_MARKERS)
def test_every_marker_is_detected_in_the_tail(marker) -> None:
bundle = _bundle(tail=f"Finished: FAILURE\n{marker.upper()}")
assert module.has_transient_infra_signature(bundle) == (True, marker)
def test_real_dns_failure_from_pip_install() -> None:
text = (
"+ pip install -r requirements.txt\n"
"WARNING: Retrying (Retry(total=4)) after connection broken by "
"'NewConnectionError(...: Failed to establish a new connection: "
"[Errno -3] Temporary failure in name resolution')\n"
"ERROR: Could not install packages due to an OSError\n"
)
matched, marker = module.has_transient_infra_signature(_bundle(regions=[text]))
assert matched is True
assert marker == "failed to establish a new connection"
def test_real_scm_connect_failure_from_git_checkout() -> None:
text = (
"+ git fetch --tags --force --progress -- https://scm.bstein.dev/bstein/titan-iac.git\n"
"fatal: unable to access 'https://scm.bstein.dev/bstein/titan-iac.git/': "
"Failed to connect to scm.bstein.dev port 443 after 130626 ms: Connection timed out\n"
)
matched, marker = module.has_transient_infra_signature(_bundle(regions=[text]))
assert matched is True
assert marker == "failed to connect to"
def test_matching_is_case_insensitive() -> None:
bundle = _bundle(regions=["Kubelet reported ErrImagePull for the agent container"])
assert module.has_transient_infra_signature(bundle) == (True, "errimagepull")
def test_earliest_region_wins_over_later_regions_and_tail() -> None:
bundle = _bundle(
regions=["stage one: connection refused", "stage two: 502 bad gateway"],
tail="Finished: FAILURE (i/o timeout)",
)
assert module.has_transient_infra_signature(bundle) == (True, "connection refused")
def test_earliest_marker_within_one_region_wins() -> None:
text = "could not resolve host: registry.example\nthen later a 503 service unavailable"
bundle = _bundle(regions=[text])
assert module.has_transient_infra_signature(bundle) == (True, "could not resolve host")
def test_regions_are_searched_before_the_tail() -> None:
bundle = _bundle(regions=["nothing infrastructural here"], tail="tls handshake timeout")
assert module.has_transient_infra_signature(bundle) == (True, "tls handshake timeout")
def test_disk_exhaustion_is_deliberately_not_transient() -> None:
text = "cp: error writing '/home/jenkins/agent/workspace/x': No space left on device"
bundle = _bundle(regions=[text], tail=text)
assert "no space left on device" not in module.INFRA_MARKERS
assert module.has_transient_infra_signature(bundle) == (False, None)
def test_ordinary_test_failure_is_not_transient() -> None:
text = "FAILED tests/test_discount.py::test_bulk - AssertionError: assert 0.9 == 0.8"
bundle = _bundle(regions=[text], tail="short test summary info")
assert module.has_transient_infra_signature(bundle) == (False, None)
@pytest.mark.parametrize(
"bundle",
[
{},
{"jenkins": None},
{"jenkins": {}},
{"jenkins": {"console_failures": None, "console_tail": None}},
{"jenkins": {"console_failures": "not-a-list", "console_tail": 7}},
{"jenkins": {"console_failures": ["not-a-dict", {"text": None}], "console_tail": ""}},
{"jenkins": {"console_failures": [{}]}},
],
)
def test_empty_or_malformed_bundles_never_match(bundle) -> None:
assert module.has_transient_infra_signature(bundle) == (False, None)
def test_non_dict_bundle_never_raises() -> None:
assert module.has_transient_infra_signature(None) == (False, None) # type: ignore[arg-type]
assert module.has_transient_infra_signature("connection refused") == (False, None) # type: ignore[arg-type]
def test_unreadable_bundle_is_swallowed() -> None:
class Exploding(dict):
def get(self, *args, **kwargs): # type: ignore[no-untyped-def]
raise RuntimeError("boom")
assert module.has_transient_infra_signature(Exploding()) == (False, None)
def test_markers_are_lowercase_and_unique() -> None:
assert list(module.INFRA_MARKERS) == [marker.lower() for marker in module.INFRA_MARKERS]
assert len(set(module.INFRA_MARKERS)) == len(module.INFRA_MARKERS)

35
tests/test_settings.py
View File

@ -123,3 +123,38 @@ def test_from_env_includes_hermes_code_settings(monkeypatch) -> None:
assert cfg.hermes_code_max_changed_lines == 10
assert cfg.hermes_gitea_base_url == "https://scm.bstein.dev"
assert cfg.hermes_gitea_token == "token"
def test_from_env_includes_hermes_action_registry(monkeypatch) -> None:
monkeypatch.setenv("ARIADNE_HERMES_ALLOWED_ACTIONS", "repair_demo_fixture, retry_transient_infra")
monkeypatch.setenv("ARIADNE_HERMES_PARAMETERIZED_JOBS", "hermes-triage-demo, seeded-demo")
cfg = Settings.from_env()
assert cfg.hermes_allowed_actions == ["repair_demo_fixture", "retry_transient_infra"]
assert cfg.hermes_parameterized_jobs == ["hermes-triage-demo", "seeded-demo"]
assert cfg.hermes_action_classifications == {
"known_demo_fixture_failure": "repair_demo_fixture",
"transient_infra_failure": "retry_transient_infra",
}
def test_hermes_action_classifications_are_overridable_and_tolerant(monkeypatch) -> None:
monkeypatch.setenv(
"ARIADNE_HERMES_ACTION_CLASSIFICATIONS",
" transient_infra_failure = retry_transient_infra , broken , = , missing_value=",
)
cfg = Settings.from_env()
assert cfg.hermes_action_classifications == {"transient_infra_failure": "retry_transient_infra"}
def test_hermes_action_defaults_stay_conservative(monkeypatch) -> None:
monkeypatch.delenv("ARIADNE_HERMES_ALLOWED_ACTIONS", raising=False)
monkeypatch.delenv("ARIADNE_HERMES_PARAMETERIZED_JOBS", raising=False)
cfg = Settings.from_env()
assert cfg.hermes_allowed_actions == ["repair_demo_fixture"]
assert cfg.hermes_parameterized_jobs == ["hermes-triage-demo"]