diff --git a/ariadne/services/hermes_autotriage.py b/ariadne/services/hermes_autotriage.py index 3f87a3f..8f7f36b 100644 --- a/ariadne/services/hermes_autotriage.py +++ b/ariadne/services/hermes_autotriage.py @@ -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, } diff --git a/ariadne/services/hermes_autotriage_decision.py b/ariadne/services/hermes_autotriage_decision.py index 0820ffc..7744401 100644 --- a/ariadne/services/hermes_autotriage_decision.py +++ b/ariadne/services/hermes_autotriage_decision.py @@ -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) diff --git a/ariadne/services/hermes_autotriage_events.py b/ariadne/services/hermes_autotriage_events.py new file mode 100644 index 0000000..517c9cf --- /dev/null +++ b/ariadne/services/hermes_autotriage_events.py @@ -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 diff --git a/ariadne/services/hermes_autotriage_repair.py b/ariadne/services/hermes_autotriage_repair.py index d4f0aa8..13c22d3 100644 --- a/ariadne/services/hermes_autotriage_repair.py +++ b/ariadne/services/hermes_autotriage_repair.py @@ -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]: diff --git a/ariadne/services/hermes_infra_signals.py b/ariadne/services/hermes_infra_signals.py new file mode 100644 index 0000000..5f1908a --- /dev/null +++ b/ariadne/services/hermes_infra_signals.py @@ -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 diff --git a/ariadne/settings.py b/ariadne/settings.py index 0be6403..0280f26 100644 --- a/ariadne/settings.py +++ b/ariadne/settings.py @@ -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 diff --git a/ariadne/settings_sections.py b/ariadne/settings_sections.py index 55a66d3..462ebf0 100644 --- a/ariadne/settings_sections.py +++ b/ariadne/settings_sections.py @@ -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( diff --git a/tests/hermes_autotriage_harness.py b/tests/hermes_autotriage_harness.py new file mode 100644 index 0000000..4dfb7bd --- /dev/null +++ b/tests/hermes_autotriage_harness.py @@ -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() diff --git a/tests/test_hermes_autotriage.py b/tests/test_hermes_autotriage.py index 1095325..7791a62 100644 --- a/tests/test_hermes_autotriage.py +++ b/tests/test_hermes_autotriage.py @@ -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 diff --git a/tests/test_hermes_autotriage_decision.py b/tests/test_hermes_autotriage_decision.py index 856d0a8..58c8d6e 100644 --- a/tests/test_hermes_autotriage_decision.py +++ b/tests/test_hermes_autotriage_decision.py @@ -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'" diff --git a/tests/test_hermes_autotriage_repair.py b/tests/test_hermes_autotriage_repair.py index 8c334a4..d2d0f14 100644 --- a/tests/test_hermes_autotriage_repair.py +++ b/tests/test_hermes_autotriage_repair.py @@ -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"] diff --git a/tests/test_hermes_autotriage_retry.py b/tests/test_hermes_autotriage_retry.py new file mode 100644 index 0000000..360650d --- /dev/null +++ b/tests/test_hermes_autotriage_retry.py @@ -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 diff --git a/tests/test_hermes_infra_signals.py b/tests/test_hermes_infra_signals.py new file mode 100644 index 0000000..85b872d --- /dev/null +++ b/tests/test_hermes_infra_signals.py @@ -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) diff --git a/tests/test_settings.py b/tests/test_settings.py index 7731d10..570c533 100644 --- a/tests/test_settings.py +++ b/tests/test_settings.py @@ -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"]