From c8d9ab20152b62756f87785df2b8d0c2df078322 Mon Sep 17 00:00:00 2001 From: codex Date: Wed, 5 Aug 2026 17:06:16 -0300 Subject: [PATCH] feat(hermes-autotriage): closed-loop demo triage via Hermes Agent API Adds the automated failure-to-repair loop for the hermes-triage-demo Jenkins job: - hermes_autotriage_logs: bounded kube-* OpenSearch evidence (fixed query, incident-ID correlation with window fallback, sanitization, byte caps) - hermes_agent_client: async /v1/runs client (bearer auth, poll, auto-deny approvals, lost-run and timeout handling) - hermes_autotriage_decision: frozen response schema parser + nine-gate action authorization (allowlist, confidence, idempotency, kill switch) - hermes_autotriage_evidence: Jenkins wfapi/testReport/console bundle assembly + failure-signature detection - hermes_autotriage_repair: hardcoded repair Job executor + one rebuild with SEED_FAILURE=false - hermes_autotriage: orchestrator state machine (detected -> diagnosed -> repairing -> awaiting_rebuild -> resolved | human_required | failed), bounded-label triage metrics, incident dedupe via storage events - settings/app: ARIADNE_HERMES_* config (default off) + 1m schedule task 126 new tests; all quality gates pass locally. Co-Authored-By: Claude Fable 5 --- ariadne/app.py | 3 + ariadne/services/hermes_agent_client.py | 199 +++++++ ariadne/services/hermes_autotriage.py | 491 ++++++++++++++++++ .../services/hermes_autotriage_decision.py | 303 +++++++++++ .../services/hermes_autotriage_evidence.py | 222 ++++++++ ariadne/services/hermes_autotriage_logs.py | 277 ++++++++++ ariadne/services/hermes_autotriage_repair.py | 180 +++++++ ariadne/settings.py | 16 + ariadne/settings_sections.py | 32 ++ tests/test_hermes_agent_client.py | 371 +++++++++++++ tests/test_hermes_autotriage.py | 479 +++++++++++++++++ tests/test_hermes_autotriage_decision.py | 359 +++++++++++++ tests/test_hermes_autotriage_evidence.py | 279 ++++++++++ tests/test_hermes_autotriage_logs.py | 385 ++++++++++++++ tests/test_hermes_autotriage_repair.py | 214 ++++++++ 15 files changed, 3810 insertions(+) create mode 100644 ariadne/services/hermes_agent_client.py create mode 100644 ariadne/services/hermes_autotriage.py create mode 100644 ariadne/services/hermes_autotriage_decision.py create mode 100644 ariadne/services/hermes_autotriage_evidence.py create mode 100644 ariadne/services/hermes_autotriage_logs.py create mode 100644 ariadne/services/hermes_autotriage_repair.py create mode 100644 tests/test_hermes_agent_client.py create mode 100644 tests/test_hermes_autotriage.py create mode 100644 tests/test_hermes_autotriage_decision.py create mode 100644 tests/test_hermes_autotriage_evidence.py create mode 100644 tests/test_hermes_autotriage_logs.py create mode 100644 tests/test_hermes_autotriage_repair.py diff --git a/ariadne/app.py b/ariadne/app.py index 8eeab24..273ccbd 100644 --- a/ariadne/app.py +++ b/ariadne/app.py @@ -22,6 +22,7 @@ from .services.comms import comms from .services.firefly import firefly from .services.game_mode import game_mode from .services.game_stream_profiles import game_stream_profiles +from .services.hermes_autotriage import run_hermes_autotriage from .services.image_sweeper import image_sweeper from .services.jenkins_build_weather import collect_jenkins_build_weather from .services.jenkins_workspace_cleanup import cleanup_jenkins_workspace_storage @@ -190,6 +191,7 @@ def _startup() -> None: scheduler.add_task("schedule.jenkins_build_weather", settings.jenkins_build_weather_cron, collect_jenkins_build_weather) scheduler.add_task("schedule.jenkins_workspace_cleanup", settings.jenkins_workspace_cleanup_cron, cleanup_jenkins_workspace_storage) scheduler.add_task("schedule.testing_triage", settings.testing_triage_cron, lambda: run_testing_triage(storage)) + scheduler.add_task("schedule.hermes_autotriage", settings.hermes_autotriage_cron, lambda: run_hermes_autotriage(storage)) scheduler.add_task("schedule.vault_k8s_auth", settings.vault_k8s_auth_cron, lambda: vault.sync_k8s_auth(wait=True)) scheduler.add_task("schedule.vault_oidc", settings.vault_oidc_cron, lambda: vault.sync_oidc(wait=True)) scheduler.add_task("schedule.comms_guest_name", settings.comms_guest_name_cron, lambda: comms.run_guest_name_randomizer(wait=True)) @@ -224,6 +226,7 @@ def _startup() -> None: "jenkins_workspace_cleanup_dry_run": settings.jenkins_workspace_cleanup_dry_run, "jenkins_workspace_cleanup_max_deletions_per_run": settings.jenkins_workspace_cleanup_max_deletions_per_run, "testing_triage_cron": settings.testing_triage_cron, + "hermes_autotriage_cron": settings.hermes_autotriage_cron, "vault_k8s_auth_cron": settings.vault_k8s_auth_cron, "vault_oidc_cron": settings.vault_oidc_cron, "comms_guest_name_cron": settings.comms_guest_name_cron, diff --git a/ariadne/services/hermes_agent_client.py b/ariadne/services/hermes_agent_client.py new file mode 100644 index 0000000..b8371ce --- /dev/null +++ b/ariadne/services/hermes_agent_client.py @@ -0,0 +1,199 @@ +from __future__ import annotations + +from dataclasses import dataclass +import time +from typing import Any + +import httpx + + +HTTP_CLIENT_ERROR_FLOOR = 400 +HTTP_NOT_FOUND = 404 +HTTP_SERVER_ERROR_FLOOR = 500 + +_DEFAULT_TOTAL_TIMEOUT_SEC = 420.0 +_DEFAULT_POLL_INTERVAL_SEC = 5.0 +_DEFAULT_REQUEST_TIMEOUT_SEC = 15.0 +_START_ATTEMPTS = 2 +_TERMINAL_STATUSES = {"completed", "failed", "cancelled"} + + +@dataclass +class HermesRunResult: + """Represent one Hermes agent triage run from start to terminal state. + + Inputs: Hermes Agent API start/poll responses observed by `run_triage`. + Outputs: a stable summary for the auto-triage decision layer; `status` is + one of completed|failed|cancelled|timeout|lost|error. + """ + + status: str + output: str | None + run_id: str | None + session_id: str | None + error: str | None + duration_seconds: float + denied_approvals: int + + +@dataclass(frozen=True) +class _RunConfig: + base_url: str + api_key: str + total_timeout_seconds: float + poll_interval_seconds: float + request_timeout_seconds: float + + +def run_triage(config: dict, prompt: str) -> HermesRunResult: + """Start one Hermes agent run and poll it until a terminal state. + + Inputs: `config` with base_url/api_key plus optional total_timeout_seconds, + poll_interval_seconds, and request_timeout_seconds; the triage prompt text. + Outputs: a HermesRunResult describing the run outcome. Approval requests + are auto-denied and counted. This function never raises. + """ + + started = time.time() + cfg = _run_config(config) + try: + with httpx.Client(timeout=cfg.request_timeout_seconds) as client: + run_id, start_error = _start_run(client, cfg, prompt) + if run_id is None: + return _error_result("error", None, start_error, started, 0) + return _poll_until_done(client, cfg, run_id, started) + except Exception as exc: + return _error_result("error", None, f"unexpected_client_failure: {exc}", started, 0) + + +def _run_config(config: dict) -> _RunConfig: + return _RunConfig( + base_url=str(config.get("base_url") or "").strip().rstrip("/"), + api_key=str(config.get("api_key") or ""), + total_timeout_seconds=_positive_float(config.get("total_timeout_seconds"), _DEFAULT_TOTAL_TIMEOUT_SEC), + poll_interval_seconds=_positive_float(config.get("poll_interval_seconds"), _DEFAULT_POLL_INTERVAL_SEC), + request_timeout_seconds=_positive_float(config.get("request_timeout_seconds"), _DEFAULT_REQUEST_TIMEOUT_SEC), + ) + + +def _positive_float(value: Any, default: float) -> float: + try: + number = float(value) + except (TypeError, ValueError): + return default + return number if number > 0 else default + + +def _headers(cfg: _RunConfig) -> dict[str, str]: + return {"Authorization": f"Bearer {cfg.api_key}"} + + +def _json_payload(resp: Any) -> dict[str, Any]: + try: + payload = resp.json() + except Exception: + return {} + return payload if isinstance(payload, dict) else {} + + +def _error_message(resp: Any) -> str: + error = _json_payload(resp).get("error") + if isinstance(error, dict) and error.get("message"): + return str(error["message"]) + return str(error or "") + + +def _start_run(client: Any, cfg: _RunConfig, prompt: str) -> tuple[str | None, str | None]: + last_error = "start_failed" + for _attempt in range(_START_ATTEMPTS): + try: + resp = client.post(f"{cfg.base_url}/v1/runs", headers=_headers(cfg), json={"input": prompt}) + except Exception as exc: + last_error = f"start_request_failed: {exc}" + continue + if resp.status_code >= HTTP_SERVER_ERROR_FLOOR: + last_error = f"start_http_{resp.status_code}" + continue + if resp.status_code >= HTTP_CLIENT_ERROR_FLOOR: + return None, f"start_http_{resp.status_code}: {_error_message(resp)}" + run_id = str(_json_payload(resp).get("run_id") or "") + if run_id: + return run_id, None + return None, "start_missing_run_id" + return None, last_error + + +def _poll_until_done(client: Any, cfg: _RunConfig, run_id: str, started: float) -> HermesRunResult: + denied = 0 + deadline = started + cfg.total_timeout_seconds + while time.time() < deadline: + outcome, payload = _poll_run(client, cfg, run_id) + if outcome == "lost": + return _error_result("lost", run_id, "run_not_found", started, denied) + if outcome == "ok": + status = str(payload.get("status") or "") + if status == "waiting_for_approval": + denied += _deny_approval(client, cfg, run_id) + elif status in _TERMINAL_STATUSES: + return _terminal_result(status, payload, run_id, started, denied) + time.sleep(cfg.poll_interval_seconds) + _stop_run(client, cfg, run_id) + return _error_result("timeout", run_id, f"total_timeout_after_{cfg.total_timeout_seconds}s", started, denied) + + +def _poll_run(client: Any, cfg: _RunConfig, run_id: str) -> tuple[str, dict[str, Any]]: + try: + resp = client.get(f"{cfg.base_url}/v1/runs/{run_id}", headers=_headers(cfg)) + except Exception: + return "request_failed", {} + if resp.status_code == HTTP_NOT_FOUND: + return "lost", _json_payload(resp) + if resp.status_code >= HTTP_CLIENT_ERROR_FLOOR: + return "request_failed", {} + return "ok", _json_payload(resp) + + +def _deny_approval(client: Any, cfg: _RunConfig, run_id: str) -> int: + try: + resp = client.post( + f"{cfg.base_url}/v1/runs/{run_id}/approval", + headers=_headers(cfg), + json={"choice": "deny"}, + ) + except Exception: + return 0 + return 1 if resp.status_code < HTTP_CLIENT_ERROR_FLOOR else 0 + + +def _stop_run(client: Any, cfg: _RunConfig, run_id: str) -> None: + try: + client.post(f"{cfg.base_url}/v1/runs/{run_id}/stop", headers=_headers(cfg)) + except Exception: + return + + +def _terminal_result(status: str, payload: dict[str, Any], run_id: str, started: float, denied: int) -> HermesRunResult: + output = payload.get("output") + session = payload.get("session_id") + error = payload.get("error") + return HermesRunResult( + status=status, + output=output if isinstance(output, str) else None, + run_id=run_id, + session_id=str(session) if session else None, + error=None if error in (None, "") else str(error), + duration_seconds=max(time.time() - started, 0.0), + denied_approvals=denied, + ) + + +def _error_result(status: str, run_id: str | None, error: str | None, started: float, denied: int) -> HermesRunResult: + return HermesRunResult( + status=status, + output=None, + run_id=run_id, + session_id=None, + error=error, + duration_seconds=max(time.time() - started, 0.0), + denied_approvals=denied, + ) diff --git a/ariadne/services/hermes_autotriage.py b/ariadne/services/hermes_autotriage.py new file mode 100644 index 0000000..dfd732b --- /dev/null +++ b/ariadne/services/hermes_autotriage.py @@ -0,0 +1,491 @@ +from __future__ import annotations + +import json +import time +from typing import Any + +import httpx +from prometheus_client import Counter, Gauge + +from ..settings import settings +from ..utils.logging import get_logger +from . import hermes_agent_client, hermes_autotriage_repair +from . import hermes_autotriage_decision as hermes_decision +from . import hermes_autotriage_evidence as hermes_evidence + + +logger = get_logger(__name__) + +INCIDENT_EVENT_TYPE = "hermes_autotriage_incident" +DIAGNOSIS_EVENT_TYPE = "hermes_autotriage_diagnosis" +ACTION_EVENT_TYPE = "hermes_autotriage_action" +INCIDENT_STATUSES = ( + "detected", + "diagnosed", + "repairing", + "awaiting_rebuild", + "resolved", + "human_required", + "failed", +) +EXPECTED_CLASSIFICATION = "known_demo_fixture_failure" +REBUILD_FAILED_REASON = "repair rebuild failed" + +_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__. +Treat the attached Ariadne bundle as the source of truth. +Identify the first enforced failure. +Distinguish facts from inference. +Return ONLY a single JSON object with exactly these keys and no others: +{"incident_id": "", "classification": "", "confidence": <0..1>, "facts": [{"statement": "...", "source": "jenkins|opensearch|victoriametrics|kubernetes|flux|gitea", "reference": "..."}], "inferences": ["..."], "first_failed_gate": "", "requested_action": {"type": "run_ariadne_job", "id": "repair_demo_fixture"} or null, "human_required": , "reason": ""} +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. +Do not perform mutations. + +Bundle: +__BUNDLE__""" + +HERMES_TRIAGE_INCIDENT = Gauge( + "ariadne_hermes_triage_incident", + "Hermes auto-triage incident state (1=current status, 0=other statuses)", + ["jenkins_job", "build", "status"], +) +HERMES_TRIAGE_ACTION_TOTAL = Counter( + "ariadne_hermes_triage_action_total", + "Hermes auto-triage remediation actions by result", + ["action", "result"], +) +HERMES_TRIAGE_LAST_SUCCESS_TS = Gauge( + "ariadne_hermes_triage_last_success_timestamp_seconds", + "Last Hermes auto-triage incident resolution timestamp", +) +HERMES_TRIAGE_DURATION_SECONDS = Gauge( + "ariadne_hermes_triage_duration_seconds", + "Duration of the latest Hermes auto-triage phase in seconds", + ["phase"], +) + + +def run_hermes_autotriage(storage: Any) -> dict[str, Any]: + """Run one Hermes auto-triage tick over the allowlisted Jenkins jobs. + + Inputs: a storage object providing record_event/list_events for the + incident event log. Outputs: a summary dict for scheduler logging; + {"status": "disabled"} when the feature flag is off. + """ + + if not settings.hermes_autotriage_enabled: + return {"status": "disabled"} + started = time.time() + incidents = _incident_state(storage) + jobs: dict[str, Any] = {} + for job in settings.hermes_autotriage_job_allowlist: + jobs[job] = _process_job(storage, job, incidents) + HERMES_TRIAGE_DURATION_SECONDS.labels(phase="total").set(time.time() - started) + logger.info( + "hermes autotriage tick finished", + extra={"event": "hermes_autotriage", "status": "ok", "jobs": json.dumps(jobs, ensure_ascii=True)}, + ) + 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.""" + + last_build = _fetch_last_build(job) + if last_build is None or last_build.get("building") or last_build.get("number") is None: + return {"status": "skipped"} + result = str(last_build.get("result") or "").upper() + if result == "SUCCESS": + return _resolve_on_success(storage, job, last_build, incidents) + if result == "FAILURE": + return _handle_failure(storage, job, last_build, incidents) + return {"status": "ignored", "result": result} + + +def _fetch_last_build(job: str) -> dict[str, Any] | None: + """Fetch the job's lastBuild summary from Jenkins, or None on failure.""" + + base_url = settings.jenkins_base_url.strip().rstrip("/") + if not base_url: + return None + try: + with httpx.Client(**_jenkins_client_kwargs()) as client: + response = client.get(f"{base_url}/job/{job}/api/json", params={"tree": _LAST_BUILD_TREE}) + response.raise_for_status() + payload = response.json() + except Exception as exc: + logger.info( + "hermes autotriage jenkins fetch failed", + extra={"event": "hermes_autotriage", "status": "jenkins_error", "job": job, "detail": str(exc)}, + ) + return None + last_build = payload.get("lastBuild") if isinstance(payload, dict) else None + return last_build if isinstance(last_build, dict) else None + + +def _jenkins_client_kwargs() -> dict[str, Any]: + """Build httpx client kwargs with Ariadne's Jenkins read credential.""" + + kwargs: dict[str, Any] = { + "timeout": settings.jenkins_api_timeout_sec, + "follow_redirects": True, + } + username = settings.jenkins_api_user.strip() + token = settings.jenkins_api_token.strip() + if username and token: + kwargs["auth"] = (username, token) + return kwargs + + +def _resolve_on_success( + storage: Any, job: str, last_build: dict[str, Any], incidents: dict[str, dict[str, Any]] +) -> dict[str, Any]: + """Resolve incidents whose rebuild completed with this successful build.""" + + number = _int_value(last_build.get("number")) + resolved: list[str] = [] + for incident in incidents.values(): + if ( + incident.get("job") == job + 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}) + HERMES_TRIAGE_LAST_SUCCESS_TS.set(time.time()) + resolved.append(str(base["incident_id"])) + return {"status": "healthy", "resolved": resolved} + + +def _handle_failure( + storage: Any, job: str, last_build: dict[str, Any], incidents: dict[str, dict[str, Any]] +) -> dict[str, Any]: + """Route a terminal build failure to dedupe, rebuild-failure, or triage.""" + + number = _int_value(last_build.get("number")) + incident_id = f"{job}/{number}" + existing = incidents.get(incident_id) + if existing is not None and existing.get("status") != "detected": + return {"status": "deduped", "incident_id": incident_id} + stale = _awaiting_rebuild_incident(incidents, job, incident_id) + if stale is not None: + base = {"incident_id": incident_id, "job": job, "build_number": number} + return _mark_rebuild_failure(storage, stale, base) + return _run_pipeline(storage, incident_id, job, last_build) + + +def _awaiting_rebuild_incident( + incidents: dict[str, dict[str, Any]], job: str, exclude_id: str +) -> dict[str, Any] | None: + """Find a different incident for this job still awaiting its rebuild.""" + + for incident_id, incident in incidents.items(): + if ( + incident_id != exclude_id + and incident.get("job") == job + and incident.get("status") == "awaiting_rebuild" + ): + return incident + return None + + +def _mark_rebuild_failure( + storage: Any, stale: dict[str, Any], base: dict[str, Any] +) -> dict[str, Any]: + """Fail the incident whose rebuild broke and escalate the new failure.""" + + stale_base = _incident_base(stale) + _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}) + return { + "status": "rebuild_failed", + "incident_id": str(base["incident_id"]), + "failed_incident": str(stale_base["incident_id"]), + } + + +def _run_pipeline( + storage: Any, incident_id: str, job: str, last_build: dict[str, Any] +) -> dict[str, Any]: + """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( + storage, + base, + "detected", + {"result": str(last_build.get("result") or ""), "url": str(last_build.get("url") or "")}, + ) + phase_started = time.time() + bundle = hermes_evidence.collect_evidence(incident_id, job, last_build) + HERMES_TRIAGE_DURATION_SECONDS.labels(phase="evidence").set(time.time() - phase_started) + phase_started = time.time() + run = hermes_agent_client.run_triage(_hermes_run_config(), _build_prompt(incident_id, bundle)) + 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}) + return {"status": "human_required", "incident_id": incident_id, "reason": reason} + outcome = hermes_decision.parse_triage_response(run.output, incident_id) + allowed, reason = hermes_decision.authorize_action( + outcome, + _decision_config(), + _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), + ) + _record_diagnosis(storage, base, run, outcome, (allowed, reason)) + if outcome.valid: + _record_incident(storage, base, "diagnosed", _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}) + return {"status": "human_required", "incident_id": incident_id, "reason": reason} + return _execute_action(storage, base, outcome) + + +def _execute_action(storage: Any, base: dict[str, Any], outcome: Any) -> dict[str, Any]: + """Run the authorized repair action and request the verification rebuild.""" + + 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}) + phase_started = time.time() + repair = hermes_autotriage_repair.execute_repair( + _repair_config(), str(base["incident_id"]), _int_value(base["build_number"]) + ) + HERMES_TRIAGE_DURATION_SECONDS.labels(phase="repair").set(time.time() - phase_started) + if not repair.get("succeeded"): + return _fail_action(storage, base, action_id, str(repair.get("error") or "repair failed")) + rebuild = hermes_autotriage_repair.trigger_rebuild(settings, str(base["job"])) + if not rebuild.get("requested"): + return _fail_action(storage, base, action_id, str(rebuild.get("error") or "rebuild trigger failed")) + _record_action(storage, base, action_id, "executed", {"repair_job": repair.get("job_name")}) + _record_incident( + storage, + base, + "awaiting_rebuild", + {"action": action_id, "repair_job": repair.get("job_name")}, + ) + return { + "status": "awaiting_rebuild", + "incident_id": str(base["incident_id"]), + "repair_job": repair.get("job_name"), + } + + +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",)) + return {"status": "failed", "incident_id": str(base["incident_id"]), "reason": error} + + +def _build_prompt(incident_id: str, bundle: dict[str, Any]) -> str: + """Render the frozen triage prompt with the incident id and bundle.""" + + compact = json.dumps(bundle, separators=(",", ":"), ensure_ascii=True) + 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 _set_incident_gauge(job: str, build: str, active: set[str]) -> None: + """Set the incident gauge to 1 for active statuses and 0 for the rest.""" + + for status in INCIDENT_STATUSES: + HERMES_TRIAGE_INCIDENT.labels(jenkins_job=job, build=build, status=status).set( + 1.0 if status in active else 0.0 + ) + + +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.""" + + 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 _UNKNOWN_ACTION_LABEL + + +def _prior_action_count(storage: Any, incident_id: str) -> int: + """Count previously recorded action events for one incident.""" + + 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")), + } + + +def _hermes_run_config() -> dict[str, Any]: + """Build the config passed to the frozen Hermes agent client.""" + + return { + "base_url": settings.hermes_api_url, + "api_key": settings.hermes_api_key, + "total_timeout_seconds": settings.hermes_run_timeout_seconds, + } + + +def _decision_config() -> dict[str, Any]: + """Build the gate config passed to the frozen authorization chain.""" + + return { + "autoremediation_enabled": settings.hermes_autoremediation_enabled, + "allowed_actions": list(settings.hermes_allowed_actions), + "min_confidence": settings.hermes_min_confidence, + "expected_classification": EXPECTED_CLASSIFICATION, + "max_actions_per_incident": settings.hermes_max_actions_per_incident, + } + + +def _repair_config() -> dict[str, Any]: + """Build the config passed to the repair Job executor.""" + + return { + "namespace": settings.hermes_demo_namespace, + "fixture_pvc": settings.hermes_demo_fixture_pvc, + "image": settings.hermes_repair_image, + } + + +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_decision.py b/ariadne/services/hermes_autotriage_decision.py new file mode 100644 index 0000000..0820ffc --- /dev/null +++ b/ariadne/services/hermes_autotriage_decision.py @@ -0,0 +1,303 @@ +from __future__ import annotations + +from dataclasses import dataclass +import json +from typing import Any + + +ALLOWED_FACT_SOURCES = {"jenkins", "opensearch", "victoriametrics", "kubernetes", "flux", "gitea"} +RUN_ARIADNE_JOB_ACTION = "run_ariadne_job" +AUTHORIZED_REASON = "authorized" + +_TOP_LEVEL_KEYS = { + "incident_id", + "classification", + "confidence", + "facts", + "inferences", + "first_failed_gate", + "requested_action", + "human_required", + "reason", +} +_STRING_FIELDS = ("incident_id", "classification", "first_failed_gate", "reason") +_FACT_KEYS = {"statement", "source", "reference"} +_ACTION_KEYS = {"type", "id"} +_CONFIDENCE_MIN = 0.0 +_CONFIDENCE_MAX = 1.0 +_DEFAULT_MIN_CONFIDENCE = 1.0 +_DEFAULT_MAX_ACTIONS = 1 +_DEFAULT_EXPECTED_CLASSIFICATION = "known_demo_fixture_failure" + + +@dataclass(frozen=True) +class TriageFact: + """Represent one evidence fact cited by a Hermes triage response. + + Inputs: a validated `facts[]` entry from the frozen response schema. + Outputs: the statement plus its allowlisted source system and reference. + """ + + statement: str + source: str + reference: str + + +@dataclass(frozen=True) +class RequestedAction: + """Represent the single remediation action a triage response may request. + + Inputs: a validated `requested_action` object from the response schema. + Outputs: the action type (always run_ariadne_job) and the job identifier. + """ + + type: str + id: str + + +@dataclass(frozen=True) +class TriageDecision: + """Represent a fully validated Hermes auto-triage response. + + Inputs: the parsed top-level fields of the frozen response schema. + Outputs: typed fields safe for the authorization gate chain. + """ + + incident_id: str + classification: str + confidence: float + facts: list[TriageFact] + inferences: list[Any] + first_failed_gate: str + requested_action: RequestedAction | None + human_required: bool + reason: str + + +@dataclass(frozen=True) +class DecisionOutcome: + """Represent the result of parsing and validating a triage response. + + Inputs: raw Hermes output run through `parse_triage_response`. + Outputs: validity, the decision when valid, whether a human must review, + and a specific reject reason when invalid. + """ + + valid: bool + decision: TriageDecision | None + human_required: bool + reject_reason: str | None + + +def parse_triage_response(raw_output: str, expected_incident_id: str) -> DecisionOutcome: + """Extract, parse, and validate a Hermes triage response payload. + + Inputs: raw model output (JSON, optionally fenced or wrapped in prose) + and the incident id the response must reference. + Outputs: a DecisionOutcome; any schema violation yields valid=False with + human_required=True and a specific reject_reason. Never raises. + """ + + candidate = _extract_json_object(str(raw_output or "")) + if candidate is None: + return _rejected("no_json_object_found") + try: + payload = json.loads(candidate) + except json.JSONDecodeError as exc: + return _rejected(f"invalid_json: {exc}") + error = _validate_payload(payload, expected_incident_id) + if error: + return _rejected(error) + decision = _decision_from_payload(payload) + return DecisionOutcome( + valid=True, + decision=decision, + human_required=decision.human_required, + reject_reason=None, + ) + + +def authorize_action( # noqa: PLR0913 - gate signature is part of the frozen module contract + outcome: DecisionOutcome, + cfg: dict, + prior_action_count: int, + build_is_terminal_failure: bool, + job_allowlisted: bool, + evidence_has_signature: bool, +) -> tuple[bool, str]: + """Apply every autoremediation gate and name the first failing gate. + + Inputs: the parsed DecisionOutcome, gate config, and deterministic facts + established outside the model (prior action count, build terminality, + job allowlist membership, evidence signature presence). + Outputs: (allowed, reason); reason is "authorized" only when every gate + passes, otherwise it names the first failing gate. + """ + + if not outcome.valid or outcome.decision is None: + 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) + gates: list[tuple[bool, str]] = [ + (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}", + ), + (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}"), + ( + decision.confidence >= min_confidence, + f"confidence_below_minimum: {decision.confidence} < {min_confidence}", + ), + (evidence_has_signature, "evidence_signature_missing"), + (_int_value(prior_action_count, 0) < max_actions, "max_actions_reached"), + (bool(cfg.get("autoremediation_enabled")), "autoremediation_disabled"), + ] + for passed, reason in gates: + if not passed: + return False, reason + return True, AUTHORIZED_REASON + + +def _rejected(reason: str) -> DecisionOutcome: + return DecisionOutcome(valid=False, decision=None, human_required=True, reject_reason=reason) + + +def _extract_json_object(raw: str) -> str | None: + depth = 0 + start = -1 + in_string = False + escaped = False + for index, char in enumerate(raw): + if in_string: + if escaped: + escaped = False + elif char == "\\": + escaped = True + elif char == '"': + in_string = False + continue + if char == '"' and depth > 0: + in_string = True + elif char == "{": + if depth == 0: + start = index + depth += 1 + elif char == "}" and depth > 0: + depth -= 1 + if depth == 0: + return raw[start : index + 1] + return None + + +def _validate_payload(payload: dict[str, Any], expected_incident_id: str) -> str | None: + keys = set(payload) + missing = sorted(_TOP_LEVEL_KEYS - keys) + if missing: + return "missing_keys: " + ", ".join(missing) + extra = sorted(keys - _TOP_LEVEL_KEYS) + if extra: + return "unexpected_keys: " + ", ".join(extra) + type_error = ( + _validate_scalar_fields(payload) + or _validate_confidence(payload["confidence"]) + or _validate_facts(payload["facts"]) + or _validate_requested_action(payload["requested_action"]) + ) + if type_error: + return type_error + if payload["incident_id"] != expected_incident_id: + return f"incident_id_mismatch: got {payload['incident_id']!r} expected {expected_incident_id!r}" + return None + + +def _validate_scalar_fields(payload: dict[str, Any]) -> str | None: + for key in _STRING_FIELDS: + if not isinstance(payload[key], str): + return f"field_type_invalid: {key} must be a string" + if not isinstance(payload["human_required"], bool): + return "field_type_invalid: human_required must be a boolean" + if not isinstance(payload["inferences"], list): + return "field_type_invalid: inferences must be a list" + return None + + +def _validate_confidence(confidence: Any) -> str | None: + if isinstance(confidence, bool) or not isinstance(confidence, (int, float)): + return "field_type_invalid: confidence must be a number" + if not _CONFIDENCE_MIN <= float(confidence) <= _CONFIDENCE_MAX: + return f"confidence_out_of_range: {confidence}" + return None + + +def _validate_facts(facts: Any) -> str | None: + if not isinstance(facts, list): + return "field_type_invalid: facts must be a list" + for index, fact in enumerate(facts): + if not isinstance(fact, dict): + return f"fact_invalid: facts[{index}] must be an object" + if set(fact) != _FACT_KEYS: + return f"fact_invalid: facts[{index}] must have exactly statement, source, reference" + if not all(isinstance(fact[key], str) for key in _FACT_KEYS): + return f"fact_invalid: facts[{index}] fields must be strings" + if fact["source"] not in ALLOWED_FACT_SOURCES: + return f"fact_source_invalid: facts[{index}] source {fact['source']!r}" + return None + + +def _validate_requested_action(action: Any) -> str | None: + if action is None: + return None + if not isinstance(action, dict): + return "requested_action_invalid: must be an object or null" + if set(action) != _ACTION_KEYS: + return "requested_action_invalid: must have exactly type and id" + if action["type"] != RUN_ARIADNE_JOB_ACTION: + return f"requested_action_invalid: unsupported type {action['type']!r}" + if not isinstance(action["id"], str) or not action["id"].strip(): + return "requested_action_invalid: id must be a non-empty string" + return None + + +def _decision_from_payload(payload: dict[str, Any]) -> TriageDecision: + action = payload["requested_action"] + return TriageDecision( + incident_id=payload["incident_id"], + classification=payload["classification"], + confidence=float(payload["confidence"]), + facts=[ + TriageFact(statement=fact["statement"], source=fact["source"], reference=fact["reference"]) + for fact in payload["facts"] + ], + inferences=list(payload["inferences"]), + first_failed_gate=payload["first_failed_gate"], + requested_action=None if action is None else RequestedAction(type=action["type"], id=action["id"]), + human_required=payload["human_required"], + reason=payload["reason"], + ) + + +def _action_id(action: RequestedAction | None) -> str: + return action.id if action is not None else "" + + +def _float_value(value: Any, default: float) -> float: + try: + return float(value) + except (TypeError, ValueError): + return default + + +def _int_value(value: Any, default: int) -> int: + try: + return int(value) + except (TypeError, ValueError): + return default diff --git a/ariadne/services/hermes_autotriage_evidence.py b/ariadne/services/hermes_autotriage_evidence.py new file mode 100644 index 0000000..98b9ac9 --- /dev/null +++ b/ariadne/services/hermes_autotriage_evidence.py @@ -0,0 +1,222 @@ +from __future__ import annotations + +from datetime import datetime, timedelta, timezone +from typing import Any + +import httpx + +from ..settings import settings +from .hermes_autotriage_logs import collect_log_evidence + + +SIGNATURE_MARKER = "hermes_demo_test_failure" +SIGNATURE_TEST_NAME = "fixture-state-check" + +_CONSOLE_TAIL_LINES = 100 +_CONSOLE_TAIL_MAX_BYTES = 8192 +_MAX_FAILED_TESTS = 10 +_MAX_ERROR_DETAILS_CHARS = 2000 +_FAILED_TEST_STATUSES = {"FAILED", "REGRESSION"} +_TEST_REPORT_TREE = "suites[cases[className,name,status,errorDetails]]" +_LOG_EXTRA_NAMESPACES = ("jenkins",) + + +def collect_evidence(incident_id: str, job: str, last_build: dict) -> dict[str, Any]: + """Assemble the bounded evidence bundle for one failed Jenkins build. + + Inputs: the incident id, Jenkins job name, and the lastBuild payload + (number/result/timestamp/duration/url) already fetched by the caller. + Outputs: the frozen bundle shape with Jenkins evidence plus sanitized + OpenSearch log evidence for the build window. Every Jenkins sub-fetch + failure is tolerated (fields become None or empty); never raises. + """ + + window_start, window_end = _build_window(last_build) + jenkins: dict[str, Any] = { + "job": job, + "build_number": _int_value(last_build.get("number")), + "url": str(last_build.get("url") or ""), + "result": str(last_build.get("result") or ""), + "timestamps": {"start": window_start, "end": window_end}, + "duration_seconds": _millis_to_seconds(last_build.get("duration")), + "first_failed_stage": None, + "console_tail": None, + "failed_tests": [], + } + build_number = jenkins["build_number"] + try: + with httpx.Client(**_client_kwargs()) as client: + jenkins["first_failed_stage"] = _first_failed_stage(client, job, build_number) + jenkins["failed_tests"] = _failed_tests(client, job, build_number) + jenkins["console_tail"] = _console_tail(client, job, build_number) + except Exception: + pass + return { + "incident_id": incident_id, + "generated_at": datetime.now(timezone.utc).isoformat(), + "jenkins": jenkins, + "log_evidence": collect_log_evidence(_log_config(), incident_id, window_start, window_end), + } + + +def evidence_has_signature(bundle: dict, incident_id: str) -> bool: + """Report whether the bundle shows the demo fixture failure signature. + + Inputs: an evidence bundle from `collect_evidence` and the incident id. + Outputs: True when the fixture-state-check test failed, the console tail + contains the demo failure marker, or the log records contain both the + marker and the incident id. + """ + + jenkins = bundle.get("jenkins") if isinstance(bundle.get("jenkins"), dict) else {} + failed_tests = jenkins.get("failed_tests") if isinstance(jenkins.get("failed_tests"), list) else [] + for test in failed_tests: + if isinstance(test, dict) and test.get("name") == SIGNATURE_TEST_NAME: + return True + if SIGNATURE_MARKER in str(jenkins.get("console_tail") or ""): + return True + log_evidence = bundle.get("log_evidence") if isinstance(bundle.get("log_evidence"), dict) else {} + records = log_evidence.get("records") if isinstance(log_evidence.get("records"), list) else [] + messages = " ".join(str(record.get("message") or "") for record in records if isinstance(record, dict)) + return SIGNATURE_MARKER in messages and incident_id in messages + + +def _build_window(last_build: dict) -> tuple[str, str]: + """Return the build start/end window as ISO-8601 UTC strings.""" + + start_ms = _number(last_build.get("timestamp")) + duration_ms = _number(last_build.get("duration")) + if start_ms <= 0: + now = datetime.now(timezone.utc) + return _iso(now), _iso(now) + start = datetime.fromtimestamp(start_ms / 1000.0, tz=timezone.utc) + end = start + timedelta(milliseconds=duration_ms) if duration_ms > 0 else datetime.now(timezone.utc) + return _iso(start), _iso(end) + + +def _iso(value: datetime) -> str: + """Format an aware datetime as an ISO-8601 UTC string.""" + + return value.astimezone(timezone.utc).isoformat() + + +def _first_failed_stage(client: httpx.Client, job: str, build_number: int) -> str | None: + """Return the first FAILED pipeline stage name, or None when unknown.""" + + try: + response = client.get(f"{_base_url()}/job/{job}/{build_number}/wfapi/describe") + response.raise_for_status() + payload = response.json() + except Exception: + return None + stages = payload.get("stages") if isinstance(payload, dict) else None + for stage in stages if isinstance(stages, list) else []: + if isinstance(stage, dict) and str(stage.get("status") or "").upper() == "FAILED": + return str(stage.get("name") or "") or None + return None + + +def _failed_tests(client: httpx.Client, job: str, build_number: int) -> list[dict[str, Any]]: + """Return bounded FAILED/REGRESSION test cases, tolerating a missing report.""" + + try: + response = client.get( + f"{_base_url()}/job/{job}/{build_number}/testReport/api/json", + params={"tree": _TEST_REPORT_TREE}, + ) + response.raise_for_status() + payload = response.json() + except Exception: + return [] + suites = payload.get("suites") if isinstance(payload, dict) else None + failed: list[dict[str, Any]] = [] + for suite in suites if isinstance(suites, list) else []: + cases = suite.get("cases") if isinstance(suite, dict) else None + for case in cases if isinstance(cases, list) else []: + if not isinstance(case, dict): + continue + if str(case.get("status") or "").upper() not in _FAILED_TEST_STATUSES: + continue + failed.append(_failed_test(case)) + if len(failed) >= _MAX_FAILED_TESTS: + return failed + return failed + + +def _failed_test(case: dict[str, Any]) -> dict[str, Any]: + """Map one test-report case to the bounded failed-test shape.""" + + details = case.get("errorDetails") + return { + "name": str(case.get("name") or ""), + "className": str(case.get("className") or ""), + "errorDetails": details[:_MAX_ERROR_DETAILS_CHARS] if isinstance(details, str) else None, + } + + +def _console_tail(client: httpx.Client, job: str, build_number: int) -> str | None: + """Return the byte-capped tail of the build console log, or None.""" + + try: + response = client.get(f"{_base_url()}/job/{job}/{build_number}/consoleText") + response.raise_for_status() + text = response.text + except Exception: + return None + tail = "\n".join(text.splitlines()[-_CONSOLE_TAIL_LINES:]) + return tail[-_CONSOLE_TAIL_MAX_BYTES:] + + +def _base_url() -> str: + """Return the configured Jenkins base URL without a trailing slash.""" + + return settings.jenkins_base_url.strip().rstrip("/") + + +def _client_kwargs() -> dict[str, Any]: + """Build httpx client kwargs with Ariadne's Jenkins read credential.""" + + kwargs: dict[str, Any] = { + "timeout": settings.jenkins_api_timeout_sec, + "follow_redirects": True, + } + username = settings.jenkins_api_user.strip() + token = settings.jenkins_api_token.strip() + if username and token: + kwargs["auth"] = (username, token) + return kwargs + + +def _log_config() -> dict[str, Any]: + """Build the config passed to the frozen OpenSearch log collector.""" + + return { + "opensearch_url": settings.opensearch_url, + "namespace": settings.hermes_demo_namespace, + "extra_namespaces": list(_LOG_EXTRA_NAMESPACES), + } + + +def _millis_to_seconds(value: Any) -> float: + """Convert a millisecond value to seconds, defaulting to zero.""" + + raw = _number(value) + return raw / 1000.0 if raw > 0 else 0.0 + + +def _number(value: Any) -> float: + """Coerce a value to float, defaulting to zero.""" + + try: + return float(value) + except (TypeError, ValueError): + return 0.0 + + +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_logs.py b/ariadne/services/hermes_autotriage_logs.py new file mode 100644 index 0000000..7957fc9 --- /dev/null +++ b/ariadne/services/hermes_autotriage_logs.py @@ -0,0 +1,277 @@ +from __future__ import annotations + +from datetime import datetime, timedelta, timezone +import json +import re +from typing import Any + +import httpx + +from ..utils.logging import get_logger + + +logger = get_logger(__name__) + +_DEFAULT_MAX_RECORDS = 50 +_HARD_CAP_RECORDS = 100 +_DEFAULT_TIMEOUT_SECONDS = 5.0 +_DEFAULT_MAX_RESPONSE_BYTES = 65536 +_WINDOW_PADDING = timedelta(minutes=5) +_HTTP_OK_MIN = 200 +_HTTP_OK_MAX = 299 + +# Field names verified live against the cluster's kube-* daily indices. +_SOURCE_FIELDS = [ + "@timestamp", + "message", + "stream", + "kubernetes.namespace_name", + "kubernetes.pod_name", + "kubernetes.container_name", +] + +_REDACTED = "[REDACTED]" + +# Demo-grade sanitization rules applied in order. Each entry is a compiled +# pattern plus a replacement template that keeps the recognizable prefix and +# masks only the sensitive remainder. Ordering matters: whole-line header +# rules run before the narrower token rules so a header value is masked once. +_SANITIZE_RULES: list[tuple[re.Pattern[str], str]] = [ + # PEM private key blocks: keep BEGIN/END markers, mask the key body. + ( + re.compile( + r"(-----BEGIN [A-Z ]*PRIVATE KEY-----)[\s\S]+?(-----END [A-Z ]*PRIVATE KEY-----)" + ), + rf"\1{_REDACTED}\2", + ), + # Authorization header values, any case, rest of line. + (re.compile(r"(authorization\s*:\s*)[^\r\n]+", re.IGNORECASE), rf"\1{_REDACTED}"), + # Cookie / Set-Cookie header values, rest of line. + ( + re.compile(r"\b((?:set-)?cookie\s*:\s*)[^\r\n]+", re.IGNORECASE), + rf"\1{_REDACTED}", + ), + # Bare `Bearer ` sequences outside an Authorization header. + (re.compile(r"\b(bearer\s+)[a-z0-9._~+/=-]+", re.IGNORECASE), rf"\1{_REDACTED}"), + # key=value / key: value assignments for credential-shaped keys, including + # prefixed variants such as `access_token` and quoted JSON-style keys. + ( + re.compile( + r"([a-z0-9_.-]*(?:password|passwd|secret|token|api_key|apikey" + r"|access_key|private_key|session(?:id)?)['\"]?\s*[=:]\s*['\"]?)" + r"[^\s;,&'\"]+", + re.IGNORECASE, + ), + rf"\1{_REDACTED}", + ), +] + + +def collect_log_evidence( + config: dict, incident_id: str, window_start: str, window_end: str +) -> dict: + """Collect bounded, sanitized OpenSearch log evidence for one incident. + + Queries the kube-* daily indices for the configured namespaces over the + build window padded by five minutes on each side. A first pass correlates + on the exact incident id in `message`; if it finds nothing, a second pass + falls back to namespace-and-window matching. Never raises: failures are + reported through the `error` key of the returned shape. + """ + + bounds = _resolved_config(config) + window = _padded_window(window_start, window_end) + if window is None: + return _shaped( + window_start, window_end, "incident_id", ([], False, "invalid window timestamps") + ) + window_from, window_to = window + correlation = "incident_id" + outcome = _run_search(bounds, (window_from, window_to), incident_id) + # Fall back to the window-only pass on a clean zero-hit first pass; a + # truncated outcome means hits existed even if the byte cap kept none. + if outcome[2] is None and not outcome[0] and not outcome[1]: + correlation = "window" + outcome = _run_search(bounds, (window_from, window_to), None) + return _shaped(window_from, window_to, correlation, outcome) + + +def _shaped( + window_from: str, + window_to: str, + correlation: str, + outcome: tuple[list[dict[str, str]], bool, str | None], +) -> dict: + """Assemble the fixed response shape from a search outcome.""" + + records, truncated, error = outcome + return { + "query_window": {"from": window_from, "to": window_to, "correlation": correlation}, + "records": records, + "truncated": truncated, + "error": error, + } + + +def _resolved_config(config: dict) -> dict[str, Any]: + """Normalize caller configuration, applying defaults and hard caps.""" + + extra = config.get("extra_namespaces") or [] + namespaces = [ + name + for name in [config.get("namespace"), *extra] + if isinstance(name, str) and name + ] + max_records = int(_number(config.get("max_records"), _DEFAULT_MAX_RECORDS)) + return { + "opensearch_url": str(config.get("opensearch_url") or "").rstrip("/"), + "namespaces": namespaces, + "max_records": min(max_records, _HARD_CAP_RECORDS), + "timeout_seconds": _number(config.get("timeout_seconds"), _DEFAULT_TIMEOUT_SECONDS), + "max_response_bytes": int( + _number(config.get("max_response_bytes"), _DEFAULT_MAX_RESPONSE_BYTES) + ), + } + + +def _number(value: Any, default: float) -> float: + """Coerce a config value to float, falling back to the default.""" + + try: + return float(value) + except (TypeError, ValueError): + return default + + +def _padded_window(window_start: str, window_end: str) -> tuple[str, str] | None: + """Return the [start - 5m, end + 5m] window as ISO strings, or None.""" + + try: + start = _parse_utc(window_start) - _WINDOW_PADDING + end = _parse_utc(window_end) + _WINDOW_PADDING + except (AttributeError, TypeError, ValueError): + return None + return _iso(start), _iso(end) + + +def _parse_utc(value: str) -> datetime: + """Parse an ISO-8601 timestamp, assuming UTC when no zone is given.""" + + parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) + if parsed.tzinfo is None: + parsed = parsed.replace(tzinfo=timezone.utc) + return parsed + + +def _iso(value: datetime) -> str: + """Format an aware datetime as an ISO-8601 UTC string with a Z suffix.""" + + return value.astimezone(timezone.utc).isoformat().replace("+00:00", "Z") + + +def _search_body(bounds: dict[str, Any], window: tuple[str, str], incident_id: str | None) -> dict: + """Build the bounded kube-* search body for one correlation pass.""" + + query: dict[str, Any] = { + "bool": { + "filter": [ + {"terms": {"kubernetes.namespace_name": bounds["namespaces"]}}, + {"range": {"@timestamp": {"gte": window[0], "lte": window[1]}}}, + ] + } + } + if incident_id is not None: + query["bool"]["must"] = [{"match_phrase": {"message": incident_id}}] + return { + "size": bounds["max_records"], + "sort": [{"@timestamp": {"order": "asc"}}], + "_source": _SOURCE_FIELDS, + "query": query, + } + + +def _run_search( + bounds: dict[str, Any], window: tuple[str, str], incident_id: str | None +) -> tuple[list[dict[str, str]], bool, str | None]: + """Run one search pass, returning (records, truncated, error).""" + + url = f"{bounds['opensearch_url']}/kube-*/_search" + body = _search_body(bounds, window, incident_id) + try: + with httpx.Client(timeout=bounds["timeout_seconds"]) as client: + response = client.post(url, json=body) + if not _HTTP_OK_MIN <= response.status_code <= _HTTP_OK_MAX: + return [], False, f"opensearch http {response.status_code}" + payload = response.json() + except httpx.TimeoutException: + return [], False, "opensearch timeout" + except ValueError: + return [], False, "opensearch returned invalid json" + except Exception as exc: + logger.info( + "opensearch evidence search failed", + extra={"event": "hermes_autotriage_logs", "status": "error", "detail": str(exc)}, + ) + return [], False, f"opensearch request failed: {exc}" + return _extract_records(payload, bounds["max_response_bytes"]) + + +def _extract_records( + payload: Any, max_response_bytes: int +) -> tuple[list[dict[str, str]], bool, str | None]: + """Turn a search payload into bounded records, enforcing the byte cap.""" + + hits_obj = payload.get("hits") if isinstance(payload, dict) else None + rows = hits_obj.get("hits") if isinstance(hits_obj, dict) else None + if not isinstance(rows, list): + return [], False, "malformed opensearch payload" + records: list[dict[str, str]] = [] + truncated = False + used_bytes = 0 + for row in rows: + record = _record(row) + if record is None: + continue + size = len(json.dumps(record, separators=(",", ":")).encode("utf-8")) + if used_bytes + size > max_response_bytes: + truncated = True + break + records.append(record) + used_bytes += size + if _reported_total(hits_obj) > len(rows): + truncated = True + return records, truncated, None + + +def _reported_total(hits_obj: dict[str, Any]) -> int: + """Return the total hit count OpenSearch reported for the query.""" + + raw_total = hits_obj.get("total") + total = raw_total.get("value") if isinstance(raw_total, dict) else raw_total + return total if isinstance(total, int) else 0 + + +def _record(row: Any) -> dict[str, str] | None: + """Map one search hit to the fixed record shape, or None if unusable.""" + + source = row.get("_source") if isinstance(row, dict) else None + if not isinstance(source, dict): + return None + raw_kubernetes = source.get("kubernetes") + kubernetes = raw_kubernetes if isinstance(raw_kubernetes, dict) else {} + return { + "index": str(row.get("_index") or ""), + "timestamp": str(source.get("@timestamp") or ""), + "namespace": str(kubernetes.get("namespace_name") or ""), + "pod": str(kubernetes.get("pod_name") or ""), + "container": str(kubernetes.get("container_name") or ""), + "message": _sanitize(str(source.get("message") or "")), + } + + +def _sanitize(text: str) -> str: + """Mask credential-shaped content in a log message with [REDACTED].""" + + for pattern, replacement in _SANITIZE_RULES: + text = pattern.sub(replacement, text) + return text diff --git a/ariadne/services/hermes_autotriage_repair.py b/ariadne/services/hermes_autotriage_repair.py new file mode 100644 index 0000000..9e9c200 --- /dev/null +++ b/ariadne/services/hermes_autotriage_repair.py @@ -0,0 +1,180 @@ +from __future__ import annotations + +import json +import time +from typing import Any + +import httpx + +from ..k8s.client import get_json, post_json +from ..utils.logging import get_logger + + +logger = get_logger(__name__) + +HTTP_CREATED = 201 +HTTP_CONFLICT = 409 + +_DEFAULT_WAIT_TIMEOUT_SECONDS = 120.0 +_POLL_INTERVAL_SECONDS = 2.0 +_JOB_TTL_SECONDS = 3600 +_REPAIR_MESSAGE = "fixture state reset to healthy" + + +def execute_repair(cfg: dict, incident_id: str, build_number: int) -> dict[str, Any]: + """Run the hardcoded demo-fixture repair Job and wait for its outcome. + + Inputs: `cfg` with namespace, fixture_pvc, and image (plus optional + wait_timeout_seconds); the incident id and the failed build number that + names the Job. Outputs: {"job_name", "succeeded", "error"}; a 409 on + creation is reported as a "duplicate job" failure. Never raises. + """ + + namespace = str(cfg.get("namespace") or "") + job_name = f"hermes-demo-repair-{build_number}" + try: + post_json( + f"/apis/batch/v1/namespaces/{namespace}/jobs", + _job_payload(cfg, job_name, incident_id), + ) + except Exception as exc: + error = "duplicate job" if _status_code(exc) == HTTP_CONFLICT else f"job create failed: {exc}" + logger.info( + "hermes demo repair job creation failed", + extra={"event": "hermes_autotriage_repair", "status": "error", "job": job_name, "detail": error}, + ) + return {"job_name": job_name, "succeeded": False, "error": error} + timeout_seconds = _number(cfg.get("wait_timeout_seconds"), _DEFAULT_WAIT_TIMEOUT_SECONDS) + return _wait_for_completion(namespace, job_name, timeout_seconds) + + +def trigger_rebuild(config: Any, job: str) -> dict[str, Any]: + """Trigger a Jenkins rebuild of `job` with the failure seed disabled. + + Inputs: a settings-like object exposing jenkins_base_url, + jenkins_api_user, jenkins_api_token, and jenkins_api_timeout_sec, plus + the Jenkins job name. Outputs: {"requested", "error"}; only an HTTP 201 + from buildWithParameters counts as success. Never raises. + """ + + 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"}, + ) + except Exception as exc: + return {"requested": False, "error": f"rebuild request failed: {exc}"} + if response.status_code == HTTP_CREATED: + return {"requested": True, "error": None} + return {"requested": False, "error": f"rebuild http {response.status_code}"} + + +def _job_payload(cfg: dict, job_name: str, incident_id: str) -> dict[str, Any]: + """Build the hardcoded batch Job manifest that resets the demo fixture.""" + + marker = json.dumps( + {"event": "hermes_demo_repair", "incident_id": incident_id, "message": _REPAIR_MESSAGE}, + separators=(",", ":"), + ) + script = f"printf 'healthy' > /fixture/state && echo '{marker}'" + return { + "apiVersion": "batch/v1", + "kind": "Job", + "metadata": { + "name": job_name, + "namespace": str(cfg.get("namespace") or ""), + "labels": { + "atlas.bstein.dev/trigger": "ariadne", + "app.kubernetes.io/part-of": "hermes-triage-demo", + }, + }, + "spec": { + "backoffLimit": 0, + "ttlSecondsAfterFinished": _JOB_TTL_SECONDS, + "template": { + "metadata": {"labels": {"app.kubernetes.io/part-of": "hermes-triage-demo"}}, + "spec": { + "restartPolicy": "Never", + "containers": [ + { + "name": "repair", + "image": str(cfg.get("image") or ""), + "command": ["sh", "-c", script], + "volumeMounts": [{"name": "fixture", "mountPath": "/fixture"}], + } + ], + "volumes": [ + { + "name": "fixture", + "persistentVolumeClaim": {"claimName": str(cfg.get("fixture_pvc") or "")}, + } + ], + }, + }, + }, + } + + +def _wait_for_completion(namespace: str, job_name: str, timeout_seconds: float) -> dict[str, Any]: + """Poll the repair Job until it succeeds, fails, or times out.""" + + deadline = time.time() + timeout_seconds + while time.time() < deadline: + try: + job = get_json(f"/apis/batch/v1/namespaces/{namespace}/jobs/{job_name}") + except Exception as exc: + return {"job_name": job_name, "succeeded": False, "error": f"job status read failed: {exc}"} + status = job.get("status") if isinstance(job.get("status"), dict) else {} + if _int_value(status.get("succeeded")) > 0: + return {"job_name": job_name, "succeeded": True, "error": None} + if _int_value(status.get("failed")) > 0: + return {"job_name": job_name, "succeeded": False, "error": "repair job failed"} + time.sleep(_POLL_INTERVAL_SECONDS) + return { + "job_name": job_name, + "succeeded": False, + "error": f"repair job timeout after {timeout_seconds}s", + } + + +def _jenkins_client_kwargs(config: Any) -> dict[str, Any]: + """Build httpx client kwargs from a settings-like Jenkins credential.""" + + kwargs: dict[str, Any] = { + "timeout": _number(getattr(config, "jenkins_api_timeout_sec", None), 10.0), + "follow_redirects": True, + } + username = str(getattr(config, "jenkins_api_user", "") or "").strip() + token = str(getattr(config, "jenkins_api_token", "") or "").strip() + if username and token: + kwargs["auth"] = (username, token) + return kwargs + + +def _status_code(exc: Exception) -> int | None: + """Extract an HTTP status code from an httpx error, when present.""" + + code = getattr(getattr(exc, "response", None), "status_code", None) + return code if isinstance(code, int) else None + + +def _number(value: Any, default: float) -> float: + """Coerce a value to float, falling back to the default.""" + + try: + return float(value) + except (TypeError, ValueError): + return default + + +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/settings.py b/ariadne/settings.py index 44f3858..4c33f11 100644 --- a/ariadne/settings.py +++ b/ariadne/settings.py @@ -9,6 +9,7 @@ from .settings_sections import ( _comms_config, _firefly_config, _game_stream_config, + _hermes_autotriage_config, _image_sweeper_config, _jenkins_build_weather_config, _jenkins_workspace_cleanup_config, @@ -179,6 +180,18 @@ class Settings: testing_triage_model_url: str testing_triage_model: str testing_triage_model_timeout_sec: float + hermes_autotriage_enabled: bool + hermes_autotriage_job_allowlist: list[str] + hermes_autoremediation_enabled: bool + hermes_allowed_actions: list[str] + hermes_min_confidence: float + hermes_max_actions_per_incident: int + hermes_api_url: str + hermes_api_key: str + hermes_run_timeout_seconds: float + hermes_demo_namespace: str + hermes_demo_fixture_pvc: str + hermes_repair_image: str vaultwarden_namespace: str vaultwarden_pod_label: str @@ -271,6 +284,7 @@ class Settings: jenkins_build_weather_cron: str jenkins_workspace_cleanup_cron: str testing_triage_cron: str + hermes_autotriage_cron: str opensearch_url: str opensearch_limit_bytes: int @@ -295,6 +309,7 @@ class Settings: jenkins_build_weather_cfg = _jenkins_build_weather_config() jenkins_workspace_cleanup_cfg = _jenkins_workspace_cleanup_config() testing_triage_cfg = _testing_triage_config() + hermes_autotriage_cfg = _hermes_autotriage_config() vaultwarden_cfg = _vaultwarden_config() schedule_cfg = _schedule_config() cluster_cfg = _cluster_state_config() @@ -339,6 +354,7 @@ class Settings: **jenkins_build_weather_cfg, **jenkins_workspace_cleanup_cfg, **testing_triage_cfg, + **hermes_autotriage_cfg, **vaultwarden_cfg, **schedule_cfg, **cluster_cfg, diff --git a/ariadne/settings_sections.py b/ariadne/settings_sections.py index 880ecb7..e047d96 100644 --- a/ariadne/settings_sections.py +++ b/ariadne/settings_sections.py @@ -254,6 +254,34 @@ def _testing_triage_config() -> dict[str, Any]: } +def _hermes_autotriage_config() -> dict[str, Any]: + return { + "hermes_autotriage_enabled": _env_bool("ARIADNE_HERMES_AUTOTRIAGE_ENABLED", "false"), + "hermes_autotriage_job_allowlist": [ + item.strip() + for item in _env("ARIADNE_HERMES_AUTOTRIAGE_JOB_ALLOWLIST", "hermes-triage-demo").split(",") + if item.strip() + ], + "hermes_autoremediation_enabled": _env_bool("ARIADNE_HERMES_AUTOREMEDIATION_ENABLED", "false"), + "hermes_allowed_actions": [ + item.strip() + for item in _env("ARIADNE_HERMES_ALLOWED_ACTIONS", "repair_demo_fixture").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( + "ARIADNE_HERMES_API_URL", + "http://hermes.hermes.svc.cluster.local:8642", + ).rstrip("/"), + "hermes_api_key": _env("ARIADNE_HERMES_API_KEY", ""), + "hermes_run_timeout_seconds": _env_float("ARIADNE_HERMES_RUN_TIMEOUT_SECONDS", 420.0), + "hermes_demo_namespace": _env("ARIADNE_HERMES_DEMO_NAMESPACE", "hermes-triage-demo"), + "hermes_demo_fixture_pvc": _env("ARIADNE_HERMES_DEMO_FIXTURE_PVC", "hermes-triage-demo-fixture"), + "hermes_repair_image": _env("ARIADNE_HERMES_REPAIR_IMAGE", "busybox:1.37"), + } + + def _vaultwarden_config() -> dict[str, Any]: return { "vaultwarden_namespace": _env("VAULTWARDEN_NAMESPACE", "vaultwarden"), @@ -311,6 +339,10 @@ def _schedule_config() -> dict[str, Any]: "ARIADNE_SCHEDULE_TESTING_TRIAGE", "*/15 * * * *", ), + "hermes_autotriage_cron": _env( + "ARIADNE_SCHEDULE_HERMES_AUTOTRIAGE", + "* * * * *", + ), "wolf_oidc_cron": _env("ARIADNE_SCHEDULE_WOLF_OIDC", _env("ARIADNE_SCHEDULE_SUNSHINE_OIDC", "17 */6 * * *")), } diff --git a/tests/test_hermes_agent_client.py b/tests/test_hermes_agent_client.py new file mode 100644 index 0000000..c3e0090 --- /dev/null +++ b/tests/test_hermes_agent_client.py @@ -0,0 +1,371 @@ +from __future__ import annotations + +from ariadne.services import hermes_agent_client as module + + +class FakeClock: + def __init__(self) -> None: + self.now = 0.0 + self.sleeps: list[float] = [] + + def time(self) -> float: + return self.now + + def sleep(self, seconds: float) -> None: + self.sleeps.append(seconds) + self.now += seconds + + +class FakeResponse: + def __init__(self, status_code: int, payload: object = None) -> None: + self.status_code = status_code + self._payload = payload + + def json(self): # type: ignore[no-untyped-def] + if isinstance(self._payload, Exception): + raise self._payload + return self._payload + + +def _install_client(monkeypatch, posts: list, gets: list) -> dict: + calls: dict = {"init_timeout": None, "posts": [], "gets": []} + + def next_item(queue: list, label: str): # type: ignore[no-untyped-def] + assert queue, f"unexpected {label} request" + item = queue.pop(0) + if isinstance(item, Exception): + raise item + return item + + class FakeClient: + def __init__(self, *, timeout=None) -> None: # type: ignore[no-untyped-def] + calls["init_timeout"] = timeout + + def __enter__(self): + return self + + def __exit__(self, *args) -> None: # type: ignore[no-untyped-def] + return None + + def post(self, url, headers=None, json=None): # type: ignore[no-untyped-def] + calls["posts"].append((url, headers, json)) + return next_item(posts, "post") + + def get(self, url, headers=None): # type: ignore[no-untyped-def] + calls["gets"].append((url, headers)) + return next_item(gets, "get") + + monkeypatch.setattr(module.httpx, "Client", FakeClient) + return calls + + +def _install_clock(monkeypatch) -> FakeClock: + clock = FakeClock() + monkeypatch.setattr(module, "time", clock) + return clock + + +def _config(**overrides) -> dict: # type: ignore[no-untyped-def] + base = { + "base_url": "http://hermes.hermes.svc.cluster.local:8642", + "api_key": "key-123", + "total_timeout_seconds": 30.0, + "poll_interval_seconds": 1.0, + "request_timeout_seconds": 7.0, + } + base.update(overrides) + return base + + +def _started(run_id: str = "run_" + "a" * 32) -> FakeResponse: + return FakeResponse(202, {"run_id": run_id, "status": "started"}) + + +def test_run_triage_completes_happy_path(monkeypatch) -> None: + _install_clock(monkeypatch) + calls = _install_client( + monkeypatch, + posts=[_started("run_abc")], + gets=[ + FakeResponse(200, {"object": "hermes.run", "status": "running"}), + FakeResponse( + 200, + { + "object": "hermes.run", + "run_id": "run_abc", + "status": "completed", + "session_id": "sess-1", + "output": "Demo fixture failure confirmed.", + "usage": {"total_tokens": 10}, + "error": None, + }, + ), + ], + ) + + result = module.run_triage(_config(), "triage incident inc-42") + + assert result.status == "completed" + assert result.output == "Demo fixture failure confirmed." + assert result.run_id == "run_abc" + assert result.session_id == "sess-1" + assert result.error is None + assert result.denied_approvals == 0 + assert result.duration_seconds == 1.0 + assert calls["init_timeout"] == 7.0 + start_url, start_headers, start_body = calls["posts"][0] + assert start_url == "http://hermes.hermes.svc.cluster.local:8642/v1/runs" + assert start_headers == {"Authorization": "Bearer key-123"} + assert start_body == {"input": "triage incident inc-42"} + poll_url, poll_headers = calls["gets"][0] + assert poll_url == "http://hermes.hermes.svc.cluster.local:8642/v1/runs/run_abc" + assert poll_headers == {"Authorization": "Bearer key-123"} + + +def test_run_triage_denies_approvals_and_keeps_polling(monkeypatch) -> None: + _install_clock(monkeypatch) + waiting = {"status": "waiting_for_approval", "last_event": "approval_requested"} + calls = _install_client( + monkeypatch, + posts=[ + _started("run_abc"), + FakeResponse(200, {"ok": True}), + ConnectionError("deny endpoint unreachable"), + FakeResponse(409, {"error": {"message": "already resolved"}}), + ], + gets=[ + FakeResponse(200, dict(waiting)), + FakeResponse(200, dict(waiting)), + FakeResponse(200, dict(waiting)), + FakeResponse(200, {"status": "completed", "output": "done", "session_id": "sess-2"}), + ], + ) + + result = module.run_triage(_config(), "prompt") + + assert result.status == "completed" + assert result.denied_approvals == 1 + deny_url, _headers, deny_body = calls["posts"][1] + assert deny_url.endswith("/v1/runs/run_abc/approval") + assert deny_body == {"choice": "deny"} + + +def test_run_triage_marks_lost_on_poll_404(monkeypatch) -> None: + _install_clock(monkeypatch) + _install_client( + monkeypatch, + posts=[_started("run_abc")], + gets=[FakeResponse(404, {"error": {"code": "run_not_found", "message": "unknown run"}})], + ) + + result = module.run_triage(_config(), "prompt") + + assert result.status == "lost" + assert result.run_id == "run_abc" + assert result.error == "run_not_found" + assert result.output is None + + +def test_run_triage_times_out_and_stops_run(monkeypatch) -> None: + clock = _install_clock(monkeypatch) + calls = _install_client( + monkeypatch, + posts=[_started("run_abc"), FakeResponse(200, {"status": "cancelling"})], + gets=[ + FakeResponse(200, {"status": "running"}), + FakeResponse(200, {"status": "running"}), + ], + ) + + result = module.run_triage(_config(total_timeout_seconds=2.0), "prompt") + + assert result.status == "timeout" + assert result.run_id == "run_abc" + assert "total_timeout_after_2.0s" in (result.error or "") + stop_url, stop_headers, _body = calls["posts"][-1] + assert stop_url.endswith("/v1/runs/run_abc/stop") + assert stop_headers == {"Authorization": "Bearer key-123"} + assert clock.sleeps == [1.0, 1.0] + + +def test_run_triage_timeout_tolerates_stop_failure(monkeypatch) -> None: + _install_clock(monkeypatch) + _install_client( + monkeypatch, + posts=[_started("run_abc"), ConnectionError("stop unreachable")], + gets=[FakeResponse(200, {"status": "queued"})], + ) + + result = module.run_triage(_config(total_timeout_seconds=1.0), "prompt") + + assert result.status == "timeout" + assert result.run_id == "run_abc" + + +def test_run_triage_retries_start_once_on_connection_error(monkeypatch) -> None: + _install_clock(monkeypatch) + calls = _install_client( + monkeypatch, + posts=[ConnectionError("connection refused"), _started("run_abc")], + gets=[FakeResponse(200, {"status": "completed", "output": "ok"})], + ) + + result = module.run_triage(_config(), "prompt") + + assert result.status == "completed" + assert len(calls["posts"]) == 2 + + +def test_run_triage_retries_start_once_on_server_error(monkeypatch) -> None: + _install_clock(monkeypatch) + calls = _install_client( + monkeypatch, + posts=[FakeResponse(503, None), _started("run_abc")], + gets=[FakeResponse(200, {"status": "completed", "output": "ok"})], + ) + + result = module.run_triage(_config(), "prompt") + + assert result.status == "completed" + assert len(calls["posts"]) == 2 + + +def test_run_triage_start_failure_after_retry_returns_error(monkeypatch) -> None: + _install_clock(monkeypatch) + calls = _install_client( + monkeypatch, + posts=[FakeResponse(500, None), ConnectionError("still down")], + gets=[], + ) + + result = module.run_triage(_config(), "prompt") + + assert result.status == "error" + assert result.run_id is None + assert "start_request_failed: still down" in (result.error or "") + assert len(calls["posts"]) == 2 + assert calls["gets"] == [] + + +def test_run_triage_does_not_retry_auth_failure(monkeypatch) -> None: + _install_clock(monkeypatch) + calls = _install_client( + monkeypatch, + posts=[FakeResponse(401, {"error": {"message": "invalid api key", "type": "invalid_request_error"}})], + gets=[], + ) + + result = module.run_triage(_config(), "prompt") + + assert result.status == "error" + assert result.error == "start_http_401: invalid api key" + assert len(calls["posts"]) == 1 + + +def test_run_triage_start_missing_run_id_returns_error(monkeypatch) -> None: + _install_clock(monkeypatch) + _install_client(monkeypatch, posts=[FakeResponse(202, {"status": "started"})], gets=[]) + + result = module.run_triage(_config(), "prompt") + + assert result.status == "error" + assert result.error == "start_missing_run_id" + + +def test_run_triage_tolerates_individual_poll_failures(monkeypatch) -> None: + clock = _install_clock(monkeypatch) + _install_client( + monkeypatch, + posts=[_started("run_abc")], + gets=[ + ConnectionError("poll dropped"), + FakeResponse(500, None), + FakeResponse(200, ValueError("bad json body")), + FakeResponse(200, ["not", "a", "dict"]), + FakeResponse(200, {"status": "completed", "output": "recovered"}), + ], + ) + + result = module.run_triage(_config(), "prompt") + + assert result.status == "completed" + assert result.output == "recovered" + assert clock.sleeps == [1.0, 1.0, 1.0, 1.0] + + +def test_run_triage_reports_failed_run_error(monkeypatch) -> None: + _install_clock(monkeypatch) + _install_client( + monkeypatch, + posts=[_started("run_abc")], + gets=[FakeResponse(200, {"status": "failed", "output": None, "error": "tool exploded"})], + ) + + result = module.run_triage(_config(), "prompt") + + assert result.status == "failed" + assert result.output is None + assert result.error == "tool exploded" + assert result.session_id is None + + +def test_run_triage_reports_cancelled_run(monkeypatch) -> None: + _install_clock(monkeypatch) + _install_client( + monkeypatch, + posts=[_started("run_abc")], + gets=[FakeResponse(200, {"status": "cancelled", "output": "partial notes", "error": None})], + ) + + result = module.run_triage(_config(), "prompt") + + assert result.status == "cancelled" + assert result.output == "partial notes" + assert result.error is None + + +def test_run_triage_never_raises_on_client_setup_failure(monkeypatch) -> None: + _install_clock(monkeypatch) + + class BoomClient: + def __init__(self, *, timeout=None) -> None: # type: ignore[no-untyped-def] + raise RuntimeError("no transport available") + + monkeypatch.setattr(module.httpx, "Client", BoomClient) + + result = module.run_triage(_config(), "prompt") + + assert result.status == "error" + assert "unexpected_client_failure: no transport available" in (result.error or "") + assert result.denied_approvals == 0 + + +def test_run_config_applies_defaults_and_normalization() -> None: + defaults = module._run_config({}) # noqa: SLF001 + assert defaults.base_url == "" + assert defaults.api_key == "" + assert defaults.total_timeout_seconds == 420.0 + assert defaults.poll_interval_seconds == 5.0 + assert defaults.request_timeout_seconds == 15.0 + + parsed = module._run_config( # noqa: SLF001 + { + "base_url": " http://hermes:8642/ ", + "api_key": "abc", + "total_timeout_seconds": "60", + "poll_interval_seconds": 0, + "request_timeout_seconds": "bad", + } + ) + assert parsed.base_url == "http://hermes:8642" + assert parsed.total_timeout_seconds == 60.0 + assert parsed.poll_interval_seconds == 5.0 + assert parsed.request_timeout_seconds == 15.0 + + +def test_payload_and_error_message_helpers() -> None: + assert module._json_payload(FakeResponse(200, ["list"])) == {} # noqa: SLF001 + assert module._json_payload(FakeResponse(200, ValueError("nope"))) == {} # noqa: SLF001 + assert module._error_message(FakeResponse(401, {"error": "denied"})) == "denied" # noqa: SLF001 + assert module._error_message(FakeResponse(401, {"error": {"message": "bad key"}})) == "bad key" # noqa: SLF001 + assert module._error_message(FakeResponse(401, {})) == "" # noqa: SLF001 diff --git a/tests/test_hermes_autotriage.py b/tests/test_hermes_autotriage.py new file mode 100644 index 0000000..81cfc2c --- /dev/null +++ b/tests/test_hermes_autotriage.py @@ -0,0 +1,479 @@ +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_pvc": "hermes-triage-demo-fixture", + "hermes_repair_image": "busybox:1.37", + "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() + + +def test_disabled_tick(monkeypatch) -> None: + env = _prepare(monkeypatch, cfg=_settings(hermes_autotriage_enabled=False)) + assert module.run_hermes_autotriage(env.storage) == {"status": "disabled"} + assert env.storage.events == [] + assert env.calls["gets"] == [] + + +def test_healthy_tick_without_incidents(monkeypatch) -> None: + env = _prepare(monkeypatch, last_build=_build(13, "SUCCESS")) + summary = module.run_hermes_autotriage(env.storage) + assert summary["status"] == "ok" + assert summary["jobs"][JOB] == {"status": "healthy", "resolved": []} + assert env.storage.events == [] + assert env.calls["triage"] == [] + + +def test_success_resolves_only_older_awaiting_rebuild(monkeypatch) -> None: + env = _prepare(monkeypatch, last_build=_build(13, "SUCCESS")) + _seed_incident(env.storage, "awaiting_rebuild", build_number=12) + _seed_incident(env.storage, "awaiting_rebuild", build_number=13) + summary = module.run_hermes_autotriage(env.storage) + assert summary["jobs"][JOB]["resolved"] == [INCIDENT_ID] + resolved = _events(env.storage, module.INCIDENT_EVENT_TYPE)[-1] + assert resolved["incident_id"] == INCIDENT_ID + assert resolved["status"] == "resolved" + assert resolved["phase"] == {"resolved_by_build": 13} + assert _gauge("12", "resolved") == 1.0 + assert _gauge("12", "human_required") == 0.0 + assert module.HERMES_TRIAGE_LAST_SUCCESS_TS._value.get() > 0 + + +def test_new_failure_full_happy_path(monkeypatch) -> None: + success_before = _counter("repair_demo_fixture", "success") + env = _prepare(monkeypatch) + summary = module.run_hermes_autotriage(env.storage) + job_summary = summary["jobs"][JOB] + assert job_summary["status"] == "awaiting_rebuild" + assert job_summary["repair_job"] == "hermes-demo-repair-12" + assert _statuses(env.storage) == ["detected", "diagnosed", "repairing", "awaiting_rebuild"] + actions = _events(env.storage, module.ACTION_EVENT_TYPE) + assert [action["result"] for action in actions] == ["requested", "accepted", "executed"] + assert all(action["action"] == "repair_demo_fixture" for action in actions) + diagnosis = _events(env.storage, module.DIAGNOSIS_EVENT_TYPE)[0] + assert diagnosis["authorized"] is True + assert diagnosis["authorize_reason"] == "authorized" + assert diagnosis["run"] == { + "status": "completed", + "run_id": "run-1", + "session_id": "sess-1", + "error": None, + "duration_seconds": 1.5, + "denied_approvals": 0, + } + assert diagnosis["outcome"]["classification"] == "known_demo_fixture_failure" + assert env.calls["repairs"] == [ + ( + { + "namespace": "hermes-triage-demo", + "fixture_pvc": "hermes-triage-demo-fixture", + "image": "busybox:1.37", + }, + INCIDENT_ID, + 12, + ) + ] + assert env.calls["rebuilds"] == [JOB] + assert _counter("repair_demo_fixture", "success") == success_before + 1.0 + assert _gauge("12", "awaiting_rebuild") == 1.0 + assert _gauge("12", "detected") == 0.0 + for phase in ("evidence", "diagnosis", "repair", "total"): + assert module.HERMES_TRIAGE_DURATION_SECONDS.labels(phase=phase)._value.get() >= 0.0 + + +def test_prompt_is_frozen_shape(monkeypatch) -> None: + env = _prepare(monkeypatch) + module.run_hermes_autotriage(env.storage) + config, prompt = env.calls["triage"][0] + assert config == { + "base_url": "http://hermes:8642", + "api_key": "key", + "total_timeout_seconds": 420.0, + } + assert prompt.startswith("Use $triage-titan-test-failures.\n") + assert f"Analyze incident {INCIDENT_ID}." in prompt + assert f'""' in prompt + assert "You are diagnosing only; you do not execute anything." in prompt + assert "Set human_required to false when the evidence matches" in prompt + assert "Do not perform mutations.\n\nBundle:\n" in prompt + assert prompt.rstrip().endswith('"log_evidence":{"records":[]}}') + + +def test_observe_mode_requires_human_without_actions(monkeypatch) -> None: + rejected_before = _counter("repair_demo_fixture", "rejected") + env = _prepare(monkeypatch, cfg=_settings(hermes_autoremediation_enabled=False)) + summary = module.run_hermes_autotriage(env.storage) + assert summary["jobs"][JOB] == { + "status": "human_required", + "incident_id": INCIDENT_ID, + "reason": "autoremediation_disabled", + } + assert _statuses(env.storage) == ["detected", "diagnosed", "human_required"] + assert _events(env.storage, module.ACTION_EVENT_TYPE) == [] + assert env.calls["repairs"] == [] + assert env.calls["rebuilds"] == [] + assert _counter("repair_demo_fixture", "rejected") == rejected_before + 1.0 + + +def test_known_incident_is_deduped(monkeypatch) -> None: + env = _prepare(monkeypatch) + _seed_incident(env.storage, "human_required") + summary = module.run_hermes_autotriage(env.storage) + assert summary["jobs"][JOB] == {"status": "deduped", "incident_id": INCIDENT_ID} + assert len(env.storage.events) == 1 + assert env.calls["triage"] == [] + + +def test_incident_state_reads_json_string_detail(monkeypatch) -> None: + env = _prepare(monkeypatch) + _seed_incident(env.storage, "resolved", as_json=True) + summary = module.run_hermes_autotriage(env.storage) + assert summary["jobs"][JOB]["status"] == "deduped" + + +def test_failed_rebuild_marks_both_incidents(monkeypatch) -> None: + env = _prepare(monkeypatch, last_build=_build(13, "FAILURE")) + _seed_incident(env.storage, "awaiting_rebuild", build_number=12) + summary = module.run_hermes_autotriage(env.storage) + assert summary["jobs"][JOB] == { + "status": "rebuild_failed", + "incident_id": f"{JOB}/13", + "failed_incident": INCIDENT_ID, + } + details = _events(env.storage, module.INCIDENT_EVENT_TYPE)[1:] + assert [(d["incident_id"], d["status"]) for d in details] == [ + (INCIDENT_ID, "failed"), + (f"{JOB}/13", "human_required"), + ] + assert details[1]["phase"] == {"reason": "repair rebuild failed"} + assert env.calls["triage"] == [] + assert env.calls["repairs"] == [] + + +@pytest.mark.parametrize("status", ["timeout", "lost", "error", "failed", "cancelled"]) +def test_unfinished_hermes_run_requires_human(monkeypatch, status) -> None: + env = _prepare(monkeypatch, run=_run(status=status, error="boom")) + summary = module.run_hermes_autotriage(env.storage) + assert summary["jobs"][JOB]["reason"] == f"hermes_run_{status}" + assert _statuses(env.storage) == ["detected", "human_required"] + diagnosis = _events(env.storage, module.DIAGNOSIS_EVENT_TYPE)[0] + assert diagnosis["run"]["status"] == status + assert diagnosis["outcome"] is None + assert diagnosis["authorized"] is False + assert env.calls["repairs"] == [] + + +def test_invalid_response_requires_human(monkeypatch) -> None: + rejected_before = _counter("unknown", "rejected") + env = _prepare(monkeypatch, run=_run(output="no json here")) + summary = module.run_hermes_autotriage(env.storage) + assert summary["jobs"][JOB]["reason"].startswith("response_invalid") + assert _statuses(env.storage) == ["detected", "human_required"] + assert _counter("unknown", "rejected") == rejected_before + 1.0 + + +def test_model_human_required_is_rejected(monkeypatch) -> None: + env = _prepare(monkeypatch, run=_run(output=_model_output(human_required=True))) + summary = module.run_hermes_autotriage(env.storage) + assert summary["jobs"][JOB]["reason"] == "human_required" + assert _statuses(env.storage) == ["detected", "diagnosed", "human_required"] + assert env.calls["repairs"] == [] + + +def test_missing_signature_is_rejected(monkeypatch) -> None: + env = _prepare(monkeypatch, signature=False) + summary = module.run_hermes_autotriage(env.storage) + assert summary["jobs"][JOB]["reason"] == "evidence_signature_missing" + assert _statuses(env.storage) == ["detected", "diagnosed", "human_required"] + assert env.calls["repairs"] == [] + + +def test_non_allowlisted_action_is_rejected(monkeypatch) -> None: + rejected_before = _counter("unknown", "rejected") + output = _model_output(requested_action={"type": "run_ariadne_job", "id": "other_action"}) + env = _prepare(monkeypatch, run=_run(output=output)) + summary = module.run_hermes_autotriage(env.storage) + assert summary["jobs"][JOB]["reason"].startswith("action_not_allowlisted") + assert _counter("unknown", "rejected") == rejected_before + 1.0 + + +def test_prior_action_blocks_second_action(monkeypatch) -> None: + env = _prepare(monkeypatch) + _seed_incident(env.storage, "detected") + env.storage.record_event( + module.ACTION_EVENT_TYPE, + {"incident_id": INCIDENT_ID, "action": "repair_demo_fixture", "result": "requested"}, + ) + summary = module.run_hermes_autotriage(env.storage) + assert summary["jobs"][JOB]["reason"] == "max_actions_reached" + assert env.calls["repairs"] == [] + + +def test_repair_failure_marks_failed_and_human(monkeypatch) -> None: + failed_before = _counter("repair_demo_fixture", "failed") + env = _prepare( + monkeypatch, + repair={"job_name": "hermes-demo-repair-12", "succeeded": False, "error": "repair job failed"}, + ) + summary = module.run_hermes_autotriage(env.storage) + assert summary["jobs"][JOB] == { + "status": "failed", + "incident_id": INCIDENT_ID, + "reason": "repair job failed", + } + assert _statuses(env.storage) == ["detected", "diagnosed", "repairing", "failed"] + actions = _events(env.storage, module.ACTION_EVENT_TYPE) + assert [action["result"] for action in actions] == ["requested", "accepted", "failed"] + assert env.calls["rebuilds"] == [] + assert _counter("repair_demo_fixture", "failed") == failed_before + 1.0 + assert _gauge("12", "failed") == 1.0 + assert _gauge("12", "human_required") == 1.0 + + +def test_rebuild_trigger_failure_marks_failed(monkeypatch) -> None: + env = _prepare(monkeypatch, rebuild={"requested": False, "error": "rebuild http 500"}) + summary = module.run_hermes_autotriage(env.storage) + assert summary["jobs"][JOB]["status"] == "failed" + assert summary["jobs"][JOB]["reason"] == "rebuild http 500" + actions = _events(env.storage, module.ACTION_EVENT_TYPE) + assert [action["result"] for action in actions] == ["requested", "accepted", "failed"] + + +def test_jenkins_fetch_failure_skips_job(monkeypatch) -> None: + env = _prepare(monkeypatch, jenkins_exc=RuntimeError("boom")) + summary = module.run_hermes_autotriage(env.storage) + assert summary["jobs"][JOB] == {"status": "skipped"} + assert env.storage.events == [] + + +def test_building_build_is_skipped(monkeypatch) -> None: + env = _prepare(monkeypatch, last_build=_build(12, None, building=True)) + assert module.run_hermes_autotriage(env.storage)["jobs"][JOB] == {"status": "skipped"} + assert env.storage.events == [] + + +def test_empty_jenkins_base_url_skips(monkeypatch) -> None: + env = _prepare(monkeypatch, cfg=_settings(jenkins_base_url="")) + assert module.run_hermes_autotriage(env.storage)["jobs"][JOB] == {"status": "skipped"} + assert env.calls["gets"] == [] + + +def test_non_terminal_result_is_ignored(monkeypatch) -> None: + env = _prepare(monkeypatch, last_build=_build(12, "ABORTED")) + assert module.run_hermes_autotriage(env.storage)["jobs"][JOB] == { + "status": "ignored", + "result": "ABORTED", + } + assert env.storage.events == [] + + +def test_event_detail_tolerates_bad_payloads() -> None: + assert module._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._int_value("not-a-number") == 0 + + +def test_jenkins_request_uses_basic_auth_and_tree(monkeypatch) -> None: + env = _prepare(monkeypatch, last_build=_build(13, "SUCCESS")) + module.run_hermes_autotriage(env.storage) + assert env.calls["client_kwargs"]["auth"] == ("user", "token") + url, params = env.calls["gets"][0] + assert url == f"https://ci.example/job/{JOB}/api/json" + assert params == {"tree": "lastBuild[number,result,building,timestamp,duration,url]"} diff --git a/tests/test_hermes_autotriage_decision.py b/tests/test_hermes_autotriage_decision.py new file mode 100644 index 0000000..856d0a8 --- /dev/null +++ b/tests/test_hermes_autotriage_decision.py @@ -0,0 +1,359 @@ +from __future__ import annotations + +import json + +import pytest + +from ariadne.services import hermes_autotriage_decision as module + + +INCIDENT_ID = "inc-42" + + +def _payload(**overrides) -> dict: # type: ignore[no-untyped-def] + base = { + "incident_id": INCIDENT_ID, + "classification": "known_demo_fixture_failure", + "confidence": 0.93, + "facts": [ + { + "statement": "Build demo #12 failed in the fixture stage.", + "source": "jenkins", + "reference": "https://ci.bstein.dev/job/demo/12/", + } + ], + "inferences": ["The fixture dataset drifted."], + "first_failed_gate": "unit-tests", + "requested_action": {"type": "run_ariadne_job", "id": "demo-fixture-reset"}, + "human_required": False, + "reason": "Known fixture failure signature matched.", + } + base.update(overrides) + return base + + +def _parse(payload: dict | None = None, raw: str | None = None, incident: str = INCIDENT_ID): # type: ignore[no-untyped-def] + text = raw if raw is not None else json.dumps(payload if payload is not None else _payload()) + return module.parse_triage_response(text, incident) + + +def _cfg(**overrides) -> dict: # type: ignore[no-untyped-def] + base = { + "autoremediation_enabled": True, + "allowed_actions": ["demo-fixture-reset"], + "min_confidence": 0.8, + "expected_classification": "known_demo_fixture_failure", + "max_actions_per_incident": 1, + } + base.update(overrides) + return base + + +def _authorize(outcome=None, cfg=None, **overrides): # type: ignore[no-untyped-def] + kwargs = { + "prior_action_count": 0, + "build_is_terminal_failure": True, + "job_allowlisted": True, + "evidence_has_signature": True, + } + kwargs.update(overrides) + return module.authorize_action(outcome if outcome is not None else _parse(), cfg if cfg is not None else _cfg(), **kwargs) + + +def test_parse_valid_response() -> None: + outcome = _parse() + + assert outcome.valid is True + assert outcome.human_required is False + assert outcome.reject_reason is None + decision = outcome.decision + assert decision is not None + assert decision.incident_id == INCIDENT_ID + assert decision.classification == "known_demo_fixture_failure" + assert decision.confidence == 0.93 + assert decision.facts == [ + module.TriageFact( + statement="Build demo #12 failed in the fixture stage.", + source="jenkins", + reference="https://ci.bstein.dev/job/demo/12/", + ) + ] + assert decision.inferences == ["The fixture dataset drifted."] + assert decision.first_failed_gate == "unit-tests" + assert decision.requested_action == module.RequestedAction(type="run_ariadne_job", id="demo-fixture-reset") + assert decision.reason == "Known fixture failure signature matched." + + +def test_parse_accepts_markdown_fenced_json() -> None: + raw = "Here is my triage.\n```json\n" + json.dumps(_payload()) + "\n```\nLet me know." + + outcome = _parse(raw=raw) + + assert outcome.valid is True + assert outcome.decision is not None + assert outcome.decision.requested_action.id == "demo-fixture-reset" + + +def test_parse_accepts_prose_and_braces_inside_strings() -> None: + payload = _payload(reason='Matched the {fixture} signature with a "quoted" note and a \\ escape.') + raw = "Prose before } stray brace... " + json.dumps(payload) + ' Trailing prose {"not": "parsed"}' + + outcome = _parse(raw=raw) + + assert outcome.valid is True + assert outcome.decision is not None + assert "{fixture}" in outcome.decision.reason + + +def test_parse_accepts_null_requested_action_and_int_confidence() -> None: + outcome = _parse(_payload(requested_action=None, confidence=1)) + + assert outcome.valid is True + assert outcome.decision is not None + assert outcome.decision.requested_action is None + assert outcome.decision.confidence == 1.0 + + +def test_parse_valid_response_with_human_required_true() -> None: + outcome = _parse(_payload(human_required=True)) + + assert outcome.valid is True + assert outcome.human_required is True + assert outcome.reject_reason is None + + +@pytest.mark.parametrize( + "raw", + ["", "no braces at all", '{"incident_id": "inc-42"', "closing only }"], +) +def test_parse_rejects_missing_json_object(raw: str) -> None: + outcome = _parse(raw=raw) + + assert outcome.valid is False + assert outcome.decision is None + assert outcome.human_required is True + assert outcome.reject_reason == "no_json_object_found" + + +def test_parse_rejects_invalid_json() -> None: + outcome = _parse(raw="{bad json}") + + assert outcome.valid is False + assert outcome.human_required is True + assert outcome.reject_reason is not None + assert outcome.reject_reason.startswith("invalid_json:") + + +def test_parse_rejects_missing_and_extra_keys() -> None: + missing = _payload() + missing.pop("reason") + outcome = _parse(missing) + assert outcome.valid is False + assert outcome.reject_reason == "missing_keys: reason" + + outcome = _parse(_payload(notes="extra")) + assert outcome.valid is False + assert outcome.reject_reason == "unexpected_keys: notes" + + +@pytest.mark.parametrize("field", ["incident_id", "classification", "first_failed_gate", "reason"]) +def test_parse_rejects_non_string_fields(field: str) -> None: + outcome = _parse(_payload(**{field: 7})) + + assert outcome.valid is False + assert outcome.reject_reason == f"field_type_invalid: {field} must be a string" + + +def test_parse_rejects_non_boolean_human_required_and_non_list_inferences() -> None: + outcome = _parse(_payload(human_required="yes")) + assert outcome.reject_reason == "field_type_invalid: human_required must be a boolean" + + outcome = _parse(_payload(inferences={"not": "a list"})) + assert outcome.reject_reason == "field_type_invalid: inferences must be a list" + + +@pytest.mark.parametrize("confidence", ["high", None, True]) +def test_parse_rejects_non_numeric_confidence(confidence) -> None: # type: ignore[no-untyped-def] + outcome = _parse(_payload(confidence=confidence)) + + assert outcome.valid is False + assert outcome.reject_reason == "field_type_invalid: confidence must be a number" + + +@pytest.mark.parametrize("confidence", [1.5, -0.2]) +def test_parse_rejects_out_of_range_confidence(confidence: float) -> None: + outcome = _parse(_payload(confidence=confidence)) + + assert outcome.valid is False + assert outcome.reject_reason == f"confidence_out_of_range: {confidence}" + + +def test_parse_rejects_bad_facts() -> None: + outcome = _parse(_payload(facts={"not": "a list"})) + assert outcome.reject_reason == "field_type_invalid: facts must be a list" + + outcome = _parse(_payload(facts=["not a dict"])) + assert outcome.reject_reason == "fact_invalid: facts[0] must be an object" + + outcome = _parse(_payload(facts=[{"statement": "s", "source": "jenkins"}])) + assert outcome.reject_reason == "fact_invalid: facts[0] must have exactly statement, source, reference" + + fact = {"statement": "s", "source": "jenkins", "reference": "r", "extra": "x"} + outcome = _parse(_payload(facts=[fact])) + assert outcome.reject_reason == "fact_invalid: facts[0] must have exactly statement, source, reference" + + outcome = _parse(_payload(facts=[{"statement": 5, "source": "jenkins", "reference": "r"}])) + assert outcome.reject_reason == "fact_invalid: facts[0] fields must be strings" + + +def test_parse_rejects_unlisted_fact_source() -> None: + good = _payload()["facts"][0] + bad = dict(good, source="bing") + outcome = _parse(_payload(facts=[good, bad])) + + assert outcome.valid is False + assert outcome.reject_reason == "fact_source_invalid: facts[1] source 'bing'" + + +def test_parse_rejects_malformed_requested_action() -> None: + outcome = _parse(_payload(requested_action=["run_ariadne_job"])) + assert outcome.reject_reason == "requested_action_invalid: must be an object or null" + + outcome = _parse(_payload(requested_action={"type": "run_ariadne_job"})) + assert outcome.reject_reason == "requested_action_invalid: must have exactly type and id" + + outcome = _parse(_payload(requested_action={"type": "run_ariadne_job", "id": "x", "why": "extra"})) + assert outcome.reject_reason == "requested_action_invalid: must have exactly type and id" + + outcome = _parse(_payload(requested_action={"type": "delete_pod", "id": "x"})) + assert outcome.reject_reason == "requested_action_invalid: unsupported type 'delete_pod'" + + outcome = _parse(_payload(requested_action={"type": "run_ariadne_job", "id": " "})) + assert outcome.reject_reason == "requested_action_invalid: id must be a non-empty string" + + outcome = _parse(_payload(requested_action={"type": "run_ariadne_job", "id": 7})) + assert outcome.reject_reason == "requested_action_invalid: id must be a non-empty string" + + +def test_parse_rejects_incident_id_mismatch() -> None: + outcome = _parse(incident="inc-99") + + assert outcome.valid is False + assert outcome.human_required is True + assert outcome.reject_reason == "incident_id_mismatch: got 'inc-42' expected 'inc-99'" + + +def test_authorize_all_gates_pass() -> None: + assert _authorize() == (True, "authorized") + + +def test_authorize_rejects_invalid_outcome() -> None: + allowed, reason = _authorize(outcome=_parse(raw="not json")) + + assert allowed is False + assert reason == "response_invalid: no_json_object_found" + + +def test_authorize_rejects_human_required_decision() -> None: + allowed, reason = _authorize(outcome=_parse(_payload(human_required=True))) + + assert allowed is False + assert reason == "human_required" + + +def test_authorize_requires_terminal_failed_build() -> None: + assert _authorize(build_is_terminal_failure=False) == (False, "build_not_terminal_failure") + + +def test_authorize_requires_allowlisted_job() -> None: + assert _authorize(job_allowlisted=False) == (False, "job_not_allowlisted") + + +def test_authorize_requires_expected_classification() -> None: + allowed, reason = _authorize(outcome=_parse(_payload(classification="novel_failure"))) + + assert allowed is False + assert reason == "classification_mismatch: got 'novel_failure' expected 'known_demo_fixture_failure'" + + +def test_authorize_requires_requested_action() -> None: + allowed, reason = _authorize(outcome=_parse(_payload(requested_action=None))) + + assert allowed is False + assert reason == "requested_action_missing" + + +def test_authorize_rejects_wrong_action_type_on_handcrafted_decision() -> None: + base = _parse().decision + assert base is not None + decision = module.TriageDecision( + incident_id=base.incident_id, + classification=base.classification, + confidence=base.confidence, + facts=base.facts, + inferences=base.inferences, + first_failed_gate=base.first_failed_gate, + requested_action=module.RequestedAction(type="delete_everything", id="demo-fixture-reset"), + human_required=False, + reason=base.reason, + ) + outcome = module.DecisionOutcome(valid=True, decision=decision, human_required=False, reject_reason=None) + + assert _authorize(outcome=outcome) == (False, "requested_action_type_invalid") + + +def test_authorize_requires_action_in_allowlist() -> None: + allowed, reason = _authorize(cfg=_cfg(allowed_actions=["other-job"])) + + assert allowed is False + assert reason == "action_not_allowlisted: 'demo-fixture-reset'" + + +def test_authorize_requires_minimum_confidence() -> None: + allowed, reason = _authorize(outcome=_parse(_payload(confidence=0.5))) + + assert allowed is False + assert reason == "confidence_below_minimum: 0.5 < 0.8" + + +def test_authorize_requires_evidence_signature() -> None: + assert _authorize(evidence_has_signature=False) == (False, "evidence_signature_missing") + + +def test_authorize_enforces_action_budget() -> None: + assert _authorize(prior_action_count=1) == (False, "max_actions_reached") + assert _authorize(prior_action_count=0, cfg=_cfg(max_actions_per_incident=2)) == (True, "authorized") + + +def test_authorize_requires_autoremediation_enabled() -> None: + allowed, reason = _authorize(cfg=_cfg(autoremediation_enabled=False)) + + assert allowed is False + assert reason == "autoremediation_disabled" + + +def test_authorize_names_first_failing_gate_in_order() -> None: + allowed, reason = _authorize( + outcome=_parse(_payload(human_required=True)), + cfg=_cfg(autoremediation_enabled=False), + build_is_terminal_failure=False, + evidence_has_signature=False, + ) + + assert allowed is False + assert reason == "human_required" + + +def test_authorize_with_empty_cfg_defaults_closed() -> None: + allowed, reason = _authorize(cfg={}) + + assert allowed is False + assert reason == "action_not_allowlisted: 'demo-fixture-reset'" + + +def test_authorize_falls_back_to_conservative_cfg_values() -> None: + cfg = _cfg(min_confidence="high", max_actions_per_incident="many") + allowed, reason = _authorize(cfg=cfg) + + assert allowed is False + assert reason == "confidence_below_minimum: 0.93 < 1.0" diff --git a/tests/test_hermes_autotriage_evidence.py b/tests/test_hermes_autotriage_evidence.py new file mode 100644 index 0000000..c4fc6d5 --- /dev/null +++ b/tests/test_hermes_autotriage_evidence.py @@ -0,0 +1,279 @@ +from __future__ import annotations + +from types import SimpleNamespace + +from ariadne.services import hermes_autotriage_evidence as module + + +JOB = "hermes-triage-demo" +INCIDENT_ID = f"{JOB}/12" + + +def _settings() -> SimpleNamespace: + return SimpleNamespace( + jenkins_base_url="https://ci.example", + jenkins_api_user="user", + jenkins_api_token="token", + jenkins_api_timeout_sec=5.0, + opensearch_url="http://opensearch:9200", + hermes_demo_namespace="hermes-triage-demo", + ) + + +class FakeResponse: + def __init__(self, payload=None, text="") -> None: # type: ignore[no-untyped-def] + self.payload = payload + self.text = text + + def raise_for_status(self) -> None: + return None + + def json(self): # type: ignore[no-untyped-def] + return self.payload + + +def _install(monkeypatch, routes, log_result=None) -> dict: # type: ignore[no-untyped-def] + calls: dict = {"gets": [], "log": [], "client_kwargs": None} + + 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)) + for suffix, item in routes.items(): + if suffix in url: + if isinstance(item, Exception): + raise item + return item + raise AssertionError(f"unexpected url {url}") + + def fake_logs(config, incident_id, window_start, window_end): # type: ignore[no-untyped-def] + calls["log"].append((config, incident_id, window_start, window_end)) + if log_result is not None: + return log_result + return {"query_window": {}, "records": [], "truncated": False, "error": None} + + monkeypatch.setattr(module, "settings", _settings()) + monkeypatch.setattr(module.httpx, "Client", FakeClient) + monkeypatch.setattr(module, "collect_log_evidence", fake_logs) + return calls + + +def _last_build(**overrides): # type: ignore[no-untyped-def] + payload = { + "number": 12, + "result": "FAILURE", + "building": False, + "timestamp": 1720000000000, + "duration": 60000, + "url": f"https://ci.example/job/{JOB}/12/", + } + payload.update(overrides) + return payload + + +def _all_failing_routes() -> dict: + return { + "/wfapi/describe": RuntimeError("stage fetch failed"), + "/testReport/api/json": RuntimeError("no test report"), + "/consoleText": RuntimeError("no console"), + } + + +def test_collect_evidence_full_bundle(monkeypatch) -> None: + routes = { + "/wfapi/describe": FakeResponse( + {"stages": [{"name": "Build", "status": "SUCCESS"}, {"name": "Test", "status": "FAILED"}]} + ), + "/testReport/api/json": FakeResponse( + { + "suites": [ + { + "cases": [ + { + "name": "fixture-state-check", + "className": "demo.Fixture", + "status": "FAILED", + "errorDetails": "boom", + }, + {"name": "ok-test", "className": "demo.Ok", "status": "PASSED"}, + { + "name": "flaky", + "className": "demo.Flaky", + "status": "REGRESSION", + "errorDetails": 5, + }, + ] + } + ] + } + ), + "/consoleText": FakeResponse(text="line1\nline2\nhermes_demo_test_failure seen"), + } + calls = _install(monkeypatch, routes) + bundle = module.collect_evidence(INCIDENT_ID, JOB, _last_build()) + + assert bundle["incident_id"] == INCIDENT_ID + assert bundle["generated_at"] + jenkins = bundle["jenkins"] + assert jenkins["job"] == JOB + assert jenkins["build_number"] == 12 + assert jenkins["result"] == "FAILURE" + assert jenkins["url"] == f"https://ci.example/job/{JOB}/12/" + assert jenkins["duration_seconds"] == 60.0 + assert jenkins["timestamps"] == { + "start": "2024-07-03T09:46:40+00:00", + "end": "2024-07-03T09:47:40+00:00", + } + assert jenkins["first_failed_stage"] == "Test" + assert jenkins["failed_tests"] == [ + {"name": "fixture-state-check", "className": "demo.Fixture", "errorDetails": "boom"}, + {"name": "flaky", "className": "demo.Flaky", "errorDetails": None}, + ] + assert jenkins["console_tail"] == "line1\nline2\nhermes_demo_test_failure seen" + assert bundle["log_evidence"]["records"] == [] + config, incident_id, window_start, window_end = calls["log"][0] + assert incident_id == INCIDENT_ID + assert (window_start, window_end) == (jenkins["timestamps"]["start"], jenkins["timestamps"]["end"]) + assert config == { + "opensearch_url": "http://opensearch:9200", + "namespace": "hermes-triage-demo", + "extra_namespaces": ["jenkins"], + } + assert calls["client_kwargs"]["auth"] == ("user", "token") + + +def test_collect_evidence_tolerates_all_fetch_failures(monkeypatch) -> None: + _install(monkeypatch, _all_failing_routes()) + bundle = module.collect_evidence(INCIDENT_ID, JOB, _last_build()) + jenkins = bundle["jenkins"] + assert jenkins["first_failed_stage"] is None + assert jenkins["failed_tests"] == [] + assert jenkins["console_tail"] is None + assert bundle["log_evidence"]["error"] is None + + +def test_collect_evidence_tolerates_client_failure(monkeypatch) -> None: + _install(monkeypatch, {}) + + def boom(**kwargs): # type: ignore[no-untyped-def] + raise RuntimeError("no client") + + monkeypatch.setattr(module.httpx, "Client", boom) + bundle = module.collect_evidence(INCIDENT_ID, JOB, _last_build()) + assert bundle["jenkins"]["failed_tests"] == [] + assert bundle["jenkins"]["console_tail"] is None + assert "log_evidence" in bundle + + +def test_console_tail_caps_lines_and_bytes(monkeypatch) -> None: + routes = _all_failing_routes() + routes["/consoleText"] = FakeResponse(text="\n".join(f"line-{i}" for i in range(300))) + _install(monkeypatch, routes) + tail = module.collect_evidence(INCIDENT_ID, JOB, _last_build())["jenkins"]["console_tail"] + lines = tail.splitlines() + assert len(lines) == 100 + assert lines[0] == "line-200" + assert lines[-1] == "line-299" + + routes["/consoleText"] = FakeResponse(text="x" * 20000) + _install(monkeypatch, routes) + tail = module.collect_evidence(INCIDENT_ID, JOB, _last_build())["jenkins"]["console_tail"] + assert len(tail) == 8192 + + +def test_failed_tests_are_capped_and_error_details_truncated(monkeypatch) -> None: + cases = [ + {"name": f"t{i}", "className": "demo.Case", "status": "FAILED", "errorDetails": "x" * 5000} + for i in range(15) + ] + routes = _all_failing_routes() + routes["/testReport/api/json"] = FakeResponse({"suites": [{"cases": cases}]}) + _install(monkeypatch, routes) + failed = module.collect_evidence(INCIDENT_ID, JOB, _last_build())["jenkins"]["failed_tests"] + assert len(failed) == 10 + assert all(len(test["errorDetails"]) == 2000 for test in failed) + + +def test_window_falls_back_when_timestamp_missing(monkeypatch) -> None: + _install(monkeypatch, _all_failing_routes()) + bundle = module.collect_evidence(INCIDENT_ID, JOB, _last_build(timestamp=None, duration=None)) + timestamps = bundle["jenkins"]["timestamps"] + assert timestamps["start"] == timestamps["end"] + assert bundle["jenkins"]["duration_seconds"] == 0.0 + + +def test_window_end_uses_now_when_duration_missing(monkeypatch) -> None: + _install(monkeypatch, _all_failing_routes()) + bundle = module.collect_evidence(INCIDENT_ID, JOB, _last_build(duration=0)) + timestamps = bundle["jenkins"]["timestamps"] + assert timestamps["start"] == "2024-07-03T09:46:40+00:00" + assert timestamps["end"] > timestamps["start"] + + +def test_evidence_tolerates_odd_jenkins_payloads(monkeypatch) -> None: + routes = { + "/wfapi/describe": FakeResponse({"stages": [{"name": "Build", "status": "SUCCESS"}]}), + "/testReport/api/json": FakeResponse( + {"suites": ["bad", {"cases": [123, {"name": "t", "className": "c", "status": "FAILED"}]}]} + ), + "/consoleText": RuntimeError("no console"), + } + _install(monkeypatch, routes) + bundle = module.collect_evidence(INCIDENT_ID, JOB, _last_build(number="not-a-number")) + jenkins = bundle["jenkins"] + assert jenkins["build_number"] == 0 + assert jenkins["first_failed_stage"] is None + assert jenkins["failed_tests"] == [{"name": "t", "className": "c", "errorDetails": None}] + + +def _bundle(failed_tests=(), console_tail="", records=()): # type: ignore[no-untyped-def] + return { + "jenkins": {"failed_tests": list(failed_tests), "console_tail": console_tail}, + "log_evidence": {"records": list(records)}, + } + + +def test_signature_from_failed_test_name() -> None: + bundle = _bundle(failed_tests=[{"name": "fixture-state-check", "className": "c", "errorDetails": None}]) + assert module.evidence_has_signature(bundle, INCIDENT_ID) is True + + +def test_signature_from_console_marker() -> None: + bundle = _bundle(console_tail="... hermes_demo_test_failure ...") + assert module.evidence_has_signature(bundle, INCIDENT_ID) is True + + +def test_signature_from_log_records_with_incident_id() -> None: + bundle = _bundle( + records=[ + {"message": "seed hermes_demo_test_failure marker"}, + {"message": f"incident {INCIDENT_ID} correlated"}, + ] + ) + assert module.evidence_has_signature(bundle, INCIDENT_ID) is True + + +def test_signature_marker_without_incident_id_is_not_enough() -> None: + bundle = _bundle(records=[{"message": "hermes_demo_test_failure only"}]) + assert module.evidence_has_signature(bundle, INCIDENT_ID) is False + + +def test_signature_absent() -> None: + bundle = _bundle( + failed_tests=[{"name": "other-test", "className": "c", "errorDetails": None}], + console_tail="all quiet", + records=[{"message": "normal log"}], + ) + assert module.evidence_has_signature(bundle, INCIDENT_ID) is False + + +def test_signature_tolerates_malformed_bundle() -> None: + assert module.evidence_has_signature({}, INCIDENT_ID) is False diff --git a/tests/test_hermes_autotriage_logs.py b/tests/test_hermes_autotriage_logs.py new file mode 100644 index 0000000..0bccff8 --- /dev/null +++ b/tests/test_hermes_autotriage_logs.py @@ -0,0 +1,385 @@ +from __future__ import annotations + +import json +from typing import Any + +import httpx + +from ariadne.services import hermes_autotriage_logs as logs_module + + +WINDOW_START = "2026-08-05T10:00:00Z" +WINDOW_END = "2026-08-05T10:30:00Z" +PADDED_FROM = "2026-08-05T09:55:00Z" +PADDED_TO = "2026-08-05T10:35:00Z" +INCIDENT_ID = "INC-4242" +DEFAULT_SIZE = 50 +HARD_CAP = 100 +DEFAULT_TIMEOUT = 5.0 +TWO_PASSES = 2 + + +def _config(**overrides: Any) -> dict: + values: dict[str, Any] = { + "opensearch_url": "http://opensearch:9200", + "namespace": "hermes", + "extra_namespaces": ["jenkins"], + } + values.update(overrides) + return values + + +def _hit(message: str = "ready", pod: str = "hermes-0") -> dict: + return { + "_index": "kube-2026.08.05", + "_source": { + "@timestamp": "2026-08-05T10:01:00Z", + "message": message, + "stream": "stdout", + "kubernetes": { + "namespace_name": "hermes", + "pod_name": pod, + "container_name": "app", + }, + }, + } + + +def _payload(hits: list, total: Any = None) -> dict: + total_value = {"value": len(hits)} if total is None else total + return {"hits": {"total": total_value, "hits": hits}} + + +class FakeResponse: + def __init__(self, payload: Any = None, status_code: int = 200, invalid_json: bool = False): + self.status_code = status_code + self._payload = payload + self._invalid_json = invalid_json + + def json(self): # type: ignore[no-untyped-def] + if self._invalid_json: + raise ValueError("bad json") + return self._payload + + +class FakeClient: + def __init__(self, script: list, calls: list): + self._script = script + self._calls = calls + + def __enter__(self): + return self + + def __exit__(self, *args) -> None: # type: ignore[no-untyped-def] + return None + + def post(self, url, json=None): # type: ignore[no-untyped-def] # noqa: A002 + self._calls.append((url, json)) + step = self._script.pop(0) + if isinstance(step, Exception): + raise step + return step + + +def _install(monkeypatch, script: list) -> tuple[list, list]: # type: ignore[no-untyped-def] + calls: list[tuple[str, dict]] = [] + client_kwargs: list[dict] = [] + + def factory(**kwargs): # type: ignore[no-untyped-def] + client_kwargs.append(kwargs) + return FakeClient(script, calls) + + monkeypatch.setattr(logs_module.httpx, "Client", factory) + return calls, client_kwargs + + +def test_first_pass_matches_incident_and_builds_bounded_query(monkeypatch) -> None: + calls, client_kwargs = _install( + monkeypatch, [FakeResponse(_payload([_hit(f"boom {INCIDENT_ID}")]))] + ) + + result = logs_module.collect_log_evidence( + _config(), INCIDENT_ID, WINDOW_START, WINDOW_END + ) + + assert result["query_window"] == { + "from": PADDED_FROM, + "to": PADDED_TO, + "correlation": "incident_id", + } + assert result["records"] == [ + { + "index": "kube-2026.08.05", + "timestamp": "2026-08-05T10:01:00Z", + "namespace": "hermes", + "pod": "hermes-0", + "container": "app", + "message": f"boom {INCIDENT_ID}", + } + ] + assert result["truncated"] is False + assert result["error"] is None + assert client_kwargs == [{"timeout": DEFAULT_TIMEOUT}] + url, body = calls[0] + assert url == "http://opensearch:9200/kube-*/_search" + assert body["size"] == DEFAULT_SIZE + assert body["sort"] == [{"@timestamp": {"order": "asc"}}] + assert body["_source"] == logs_module._SOURCE_FIELDS # noqa: SLF001 + assert body["query"]["bool"]["filter"] == [ + {"terms": {"kubernetes.namespace_name": ["hermes", "jenkins"]}}, + {"range": {"@timestamp": {"gte": PADDED_FROM, "lte": PADDED_TO}}}, + ] + assert body["query"]["bool"]["must"] == [{"match_phrase": {"message": INCIDENT_ID}}] + + +def test_second_pass_drops_incident_filter_when_first_is_empty(monkeypatch) -> None: + calls, _ = _install( + monkeypatch, [FakeResponse(_payload([])), FakeResponse(_payload([_hit()]))] + ) + + result = logs_module.collect_log_evidence( + _config(), INCIDENT_ID, WINDOW_START, WINDOW_END + ) + + assert len(calls) == TWO_PASSES + assert "must" in calls[0][1]["query"]["bool"] + assert "must" not in calls[1][1]["query"]["bool"] + assert result["query_window"]["correlation"] == "window" + assert result["records"][0]["pod"] == "hermes-0" + assert result["error"] is None + + +def test_empty_both_passes_is_success(monkeypatch) -> None: + calls, _ = _install( + monkeypatch, [FakeResponse(_payload([])), FakeResponse(_payload([]))] + ) + + result = logs_module.collect_log_evidence( + _config(), INCIDENT_ID, WINDOW_START, WINDOW_END + ) + + assert len(calls) == TWO_PASSES + assert result["records"] == [] + assert result["truncated"] is False + assert result["error"] is None + assert result["query_window"]["correlation"] == "window" + + +def test_timeout_returns_error_without_second_pass(monkeypatch) -> None: + calls, _ = _install(monkeypatch, [httpx.TimeoutException("slow")]) + + result = logs_module.collect_log_evidence( + _config(), INCIDENT_ID, WINDOW_START, WINDOW_END + ) + + assert len(calls) == 1 + assert result["error"] == "opensearch timeout" + assert result["records"] == [] + assert result["truncated"] is False + assert result["query_window"]["correlation"] == "incident_id" + + +def test_http_error_and_generic_failure_set_error(monkeypatch) -> None: + _install(monkeypatch, [FakeResponse(status_code=500)]) + result = logs_module.collect_log_evidence( + _config(), INCIDENT_ID, WINDOW_START, WINDOW_END + ) + assert result["error"] == "opensearch http 500" + assert result["records"] == [] + + _install(monkeypatch, [RuntimeError("connection refused")]) + result = logs_module.collect_log_evidence( + _config(), INCIDENT_ID, WINDOW_START, WINDOW_END + ) + assert result["error"] == "opensearch request failed: connection refused" + + +def test_invalid_json_and_malformed_payload_set_error(monkeypatch) -> None: + _install(monkeypatch, [FakeResponse(invalid_json=True)]) + result = logs_module.collect_log_evidence( + _config(), INCIDENT_ID, WINDOW_START, WINDOW_END + ) + assert result["error"] == "opensearch returned invalid json" + + _install(monkeypatch, [FakeResponse({"unexpected": True})]) + result = logs_module.collect_log_evidence( + _config(), INCIDENT_ID, WINDOW_START, WINDOW_END + ) + assert result["error"] == "malformed opensearch payload" + assert result["records"] == [] + + +def test_truncated_when_opensearch_reports_more_hits(monkeypatch) -> None: + _install(monkeypatch, [FakeResponse(_payload([_hit()], total={"value": 500}))]) + result = logs_module.collect_log_evidence( + _config(), INCIDENT_ID, WINDOW_START, WINDOW_END + ) + assert result["truncated"] is True + assert len(result["records"]) == 1 + assert result["error"] is None + + _install(monkeypatch, [FakeResponse(_payload([_hit()], total=7))]) + result = logs_module.collect_log_evidence( + _config(), INCIDENT_ID, WINDOW_START, WINDOW_END + ) + assert result["truncated"] is True + + +def test_truncated_by_byte_cap(monkeypatch) -> None: + record = { + "index": "kube-2026.08.05", + "timestamp": "2026-08-05T10:01:00Z", + "namespace": "hermes", + "pod": "hermes-0", + "container": "app", + "message": "ready", + } + record_bytes = len(json.dumps(record, separators=(",", ":")).encode("utf-8")) + hits = [_hit(), _hit(), _hit()] + _install(monkeypatch, [FakeResponse(_payload(hits))]) + + result = logs_module.collect_log_evidence( + _config(max_response_bytes=record_bytes + 1), + INCIDENT_ID, + WINDOW_START, + WINDOW_END, + ) + + assert result["records"] == [record] + assert result["truncated"] is True + assert result["error"] is None + + +def test_zero_byte_budget_keeps_no_records_without_fallback(monkeypatch) -> None: + calls, _ = _install(monkeypatch, [FakeResponse(_payload([_hit()]))]) + + result = logs_module.collect_log_evidence( + _config(max_response_bytes=1), INCIDENT_ID, WINDOW_START, WINDOW_END + ) + + assert len(calls) == 1 + assert result["records"] == [] + assert result["truncated"] is True + assert result["query_window"]["correlation"] == "incident_id" + + +def test_max_records_hard_cap_and_bad_values(monkeypatch) -> None: + calls, client_kwargs = _install( + monkeypatch, [FakeResponse(_payload([_hit()])), FakeResponse(_payload([_hit()]))] + ) + + logs_module.collect_log_evidence( + _config(max_records=250, timeout_seconds=2.5), + INCIDENT_ID, + WINDOW_START, + WINDOW_END, + ) + logs_module.collect_log_evidence( + _config(max_records="bad"), INCIDENT_ID, WINDOW_START, WINDOW_END + ) + + assert calls[0][1]["size"] == HARD_CAP + assert client_kwargs[0] == {"timeout": 2.5} + assert calls[1][1]["size"] == DEFAULT_SIZE + assert client_kwargs[1] == {"timeout": DEFAULT_TIMEOUT} + + +def test_invalid_window_timestamps_skip_search(monkeypatch) -> None: + calls, _ = _install(monkeypatch, []) + + result = logs_module.collect_log_evidence( + _config(), INCIDENT_ID, "not-a-time", WINDOW_END + ) + + assert calls == [] + assert result == { + "query_window": { + "from": "not-a-time", + "to": WINDOW_END, + "correlation": "incident_id", + }, + "records": [], + "truncated": False, + "error": "invalid window timestamps", + } + + +def test_window_arithmetic_pads_five_minutes_and_assumes_utc() -> None: + padded = logs_module._padded_window( # noqa: SLF001 + "2026-08-05T10:00:00", "2026-08-05T10:30:00" + ) + assert padded == (PADDED_FROM, PADDED_TO) + assert logs_module._padded_window(None, WINDOW_END) is None # noqa: SLF001 + + +def test_malformed_hits_are_skipped_and_missing_total_is_success(monkeypatch) -> None: + bad_source = {"_index": "kube-2026.08.05", "_source": "oops"} + no_kubernetes = {"_index": "kube-2026.08.05", "_source": {"message": "plain"}} + _install( + monkeypatch, + [FakeResponse({"hits": {"hits": ["junk", bad_source, no_kubernetes, _hit()]}})], + ) + + result = logs_module.collect_log_evidence( + _config(), INCIDENT_ID, WINDOW_START, WINDOW_END + ) + + assert result["error"] is None + assert result["truncated"] is False + assert [item["pod"] for item in result["records"]] == ["", "hermes-0"] + assert result["records"][0] == { + "index": "kube-2026.08.05", + "timestamp": "", + "namespace": "", + "pod": "", + "container": "", + "message": "plain", + } + + +def test_sanitize_masks_headers_and_tokens() -> None: + sanitize = logs_module._sanitize # noqa: SLF001 + + assert sanitize("Authorization: Bearer abc.def") == "Authorization: [REDACTED]" + assert sanitize("authorization:Basic Zm9vOmJhcg==") == "authorization:[REDACTED]" + assert sanitize("retry with Bearer eyJhbGci.sig") == "retry with Bearer [REDACTED]" + assert sanitize("Cookie: sid=abc; theme=dark") == "Cookie: [REDACTED]" + assert sanitize("Set-Cookie: id=1; Path=/") == "Set-Cookie: [REDACTED]" + assert sanitize("listening on port 8080") == "listening on port 8080" + + +def test_sanitize_masks_credential_assignments() -> None: + sanitize = logs_module._sanitize # noqa: SLF001 + + assert sanitize("password=hunter2 ok") == "password=[REDACTED] ok" + assert sanitize("passwd: hunter2") == "passwd: [REDACTED]" + assert sanitize("db_secret=s3cr3t") == "db_secret=[REDACTED]" + assert sanitize("token: abc123") == "token: [REDACTED]" + assert sanitize("api_key=k1") == "api_key=[REDACTED]" + assert sanitize("apikey: k2") == "apikey: [REDACTED]" + assert sanitize("access_key=AKIA123") == "access_key=[REDACTED]" + assert sanitize("private_key=xyz") == "private_key=[REDACTED]" + assert sanitize("access_token=opaque") == "access_token=[REDACTED]" + assert sanitize("session=deadbeef;") == "session=[REDACTED];" + assert sanitize("sessionid: deadbeef") == "sessionid: [REDACTED]" + assert sanitize('"password": "hunter2"') == '"password": "[REDACTED]"' + + +def test_sanitize_masks_pem_private_key_blocks() -> None: + text = ( + "before\n" + "-----BEGIN RSA PRIVATE KEY-----\n" + "MIIEowIBAAKCAQEA\nabcdef\n" + "-----END RSA PRIVATE KEY-----\n" + "after" + ) + + masked = logs_module._sanitize(text) # noqa: SLF001 + + assert masked == ( + "before\n" + "-----BEGIN RSA PRIVATE KEY-----" + "[REDACTED]" + "-----END RSA PRIVATE KEY-----\n" + "after" + ) diff --git a/tests/test_hermes_autotriage_repair.py b/tests/test_hermes_autotriage_repair.py new file mode 100644 index 0000000..bda8c3c --- /dev/null +++ b/tests/test_hermes_autotriage_repair.py @@ -0,0 +1,214 @@ +from __future__ import annotations + +from types import SimpleNamespace + +from ariadne.services import hermes_autotriage_repair as module + + +INCIDENT_ID = "hermes-triage-demo/12" + + +class FakeClock: + def __init__(self) -> None: + self.now = 0.0 + self.sleeps: list[float] = [] + + def time(self) -> float: + return self.now + + def sleep(self, seconds: float) -> None: + self.sleeps.append(seconds) + self.now += seconds + + +class FakeResponse: + def __init__(self, status_code: int) -> None: + self.status_code = status_code + + +def _cfg(**overrides) -> dict: # type: ignore[no-untyped-def] + base = { + "namespace": "hermes-triage-demo", + "fixture_pvc": "hermes-triage-demo-fixture", + "image": "busybox:1.37", + } + base.update(overrides) + return base + + +def _jenkins_settings(**overrides) -> SimpleNamespace: # type: ignore[no-untyped-def] + values = { + "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 _install_k8s(monkeypatch, statuses, post_exc=None) -> dict: # type: ignore[no-untyped-def] + calls: dict = {"posts": [], "gets": []} + queue = list(statuses) + + def fake_post(path, payload): # type: ignore[no-untyped-def] + calls["posts"].append((path, payload)) + if post_exc is not None: + raise post_exc + return {"metadata": {"name": payload["metadata"]["name"]}} + + def fake_get(path): # type: ignore[no-untyped-def] + calls["gets"].append(path) + item = queue.pop(0) if queue else {"status": {"active": 1}} + if isinstance(item, Exception): + raise item + return item + + monkeypatch.setattr(module, "post_json", fake_post) + monkeypatch.setattr(module, "get_json", fake_get) + clock = FakeClock() + monkeypatch.setattr(module, "time", clock) + calls["clock"] = clock + return calls + + +def test_repair_job_payload_contract(monkeypatch) -> None: + calls = _install_k8s(monkeypatch, [{"status": {"succeeded": 1}}]) + result = module.execute_repair(_cfg(), INCIDENT_ID, 12) + + assert result == {"job_name": "hermes-demo-repair-12", "succeeded": True, "error": None} + path, payload = calls["posts"][0] + assert path == "/apis/batch/v1/namespaces/hermes-triage-demo/jobs" + assert payload["apiVersion"] == "batch/v1" + assert payload["kind"] == "Job" + assert payload["metadata"]["name"] == "hermes-demo-repair-12" + assert payload["metadata"]["namespace"] == "hermes-triage-demo" + assert payload["metadata"]["labels"] == { + "atlas.bstein.dev/trigger": "ariadne", + "app.kubernetes.io/part-of": "hermes-triage-demo", + } + spec = payload["spec"] + assert spec["backoffLimit"] == 0 + assert spec["ttlSecondsAfterFinished"] == 3600 + pod = spec["template"]["spec"] + assert pod["restartPolicy"] == "Never" + container = pod["containers"][0] + assert container["image"] == "busybox:1.37" + assert container["command"][:2] == ["sh", "-c"] + script = container["command"][2] + assert "printf 'healthy' > /fixture/state" in script + assert '"event":"hermes_demo_repair"' in script + assert f'"incident_id":"{INCIDENT_ID}"' in script + assert '"message":"fixture state reset to healthy"' in script + assert container["volumeMounts"] == [{"name": "fixture", "mountPath": "/fixture"}] + assert pod["volumes"] == [ + {"name": "fixture", "persistentVolumeClaim": {"claimName": "hermes-triage-demo-fixture"}} + ] + + +def test_execute_repair_waits_through_active_polls(monkeypatch) -> None: + calls = _install_k8s( + monkeypatch, + [{"status": {"active": 1}}, {"status": {"succeeded": 1}}], + ) + result = module.execute_repair(_cfg(), INCIDENT_ID, 12) + assert result["succeeded"] is True + assert len(calls["gets"]) == 2 + assert calls["clock"].sleeps == [2.0] + + +def test_execute_repair_job_failure(monkeypatch) -> None: + _install_k8s(monkeypatch, [{"status": {"failed": 1}}]) + result = module.execute_repair(_cfg(), INCIDENT_ID, 12) + assert result == {"job_name": "hermes-demo-repair-12", "succeeded": False, "error": "repair job failed"} + + +def test_execute_repair_timeout(monkeypatch) -> None: + _install_k8s(monkeypatch, []) + result = module.execute_repair(_cfg(wait_timeout_seconds=6), INCIDENT_ID, 12) + assert result["succeeded"] is False + assert result["error"] == "repair job timeout after 6.0s" + + +def test_execute_repair_duplicate_job(monkeypatch) -> None: + conflict = RuntimeError("conflict") + conflict.response = SimpleNamespace(status_code=409) # type: ignore[attr-defined] + _install_k8s(monkeypatch, [], post_exc=conflict) + result = module.execute_repair(_cfg(), INCIDENT_ID, 12) + assert result == {"job_name": "hermes-demo-repair-12", "succeeded": False, "error": "duplicate job"} + + +def test_execute_repair_create_error(monkeypatch) -> None: + _install_k8s(monkeypatch, [], post_exc=ValueError("nope")) + result = module.execute_repair(_cfg(), INCIDENT_ID, 12) + assert result["succeeded"] is False + assert result["error"] == "job create failed: nope" + + +def test_execute_repair_status_read_error(monkeypatch) -> None: + _install_k8s(monkeypatch, [RuntimeError("api down")]) + result = module.execute_repair(_cfg(), INCIDENT_ID, 12) + assert result["succeeded"] is False + assert result["error"] == "job status read failed: api down" + + +def _install_http(monkeypatch, response=None, exc=None) -> dict: # type: ignore[no-untyped-def] + calls: dict = {"posts": [], "kwargs": None} + + class FakeClient: + def __init__(self, **kwargs) -> None: # type: ignore[no-untyped-def] + calls["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 post(self, url, data=None): # type: ignore[no-untyped-def] + calls["posts"].append((url, data)) + if exc is not None: + raise exc + return response + + monkeypatch.setattr(module.httpx, "Client", FakeClient) + return calls + + +def test_trigger_rebuild_success(monkeypatch) -> None: + calls = _install_http(monkeypatch, response=FakeResponse(201)) + result = module.trigger_rebuild(_jenkins_settings(), "hermes-triage-demo") + 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") + assert calls["kwargs"]["timeout"] == 5.0 + + +def test_trigger_rebuild_non_created_status(monkeypatch) -> None: + _install_http(monkeypatch, response=FakeResponse(500)) + result = module.trigger_rebuild(_jenkins_settings(), "hermes-triage-demo") + assert result == {"requested": False, "error": "rebuild http 500"} + + +def test_trigger_rebuild_request_failure(monkeypatch) -> None: + _install_http(monkeypatch, exc=RuntimeError("connect refused")) + result = module.trigger_rebuild(_jenkins_settings(), "hermes-triage-demo") + assert result["requested"] is False + assert result["error"] == "rebuild request failed: connect refused" + + +def test_trigger_rebuild_without_base_url(monkeypatch) -> None: + calls = _install_http(monkeypatch, response=FakeResponse(201)) + result = module.trigger_rebuild(_jenkins_settings(jenkins_base_url=""), "hermes-triage-demo") + assert result == {"requested": False, "error": "jenkins base url is empty"} + assert calls["posts"] == [] + + +def test_trigger_rebuild_without_credentials(monkeypatch) -> None: + calls = _install_http(monkeypatch, response=FakeResponse(201)) + config = _jenkins_settings(jenkins_api_user="", jenkins_api_token="") + result = module.trigger_rebuild(config, "hermes-triage-demo") + assert result["requested"] is True + assert "auth" not in calls["kwargs"]