diff --git a/alert-processor/app/matrix_notifier.py b/alert-processor/app/matrix_notifier.py new file mode 100644 index 0000000..0652f8a --- /dev/null +++ b/alert-processor/app/matrix_notifier.py @@ -0,0 +1,220 @@ +from __future__ import annotations + +import mimetypes +import uuid +from dataclasses import dataclass +from pathlib import Path +from urllib.parse import quote + +import httpx + +from app.matrix_token_manager import MatrixTokenManager + + +@dataclass +class MatrixSendResult: + ok: bool + status_code: int | None + event_id: str | None + error: str | None + content_uri: str | None = None + + +@dataclass +class MatrixUploadResult: + ok: bool + status_code: int | None + content_uri: str | None + error: str | None + + +class MatrixNotifier: + def __init__( + self, + homeserver_url: str, + room_id: str, + token_manager: MatrixTokenManager, + message_type: str = "m.notice", + timeout_seconds: float = 10, + verify_tls: bool = True, + ) -> None: + self.homeserver_url = homeserver_url.rstrip("/") + self.room_id = room_id + self.token_manager = token_manager + self.message_type = message_type + self.client = httpx.AsyncClient( + timeout=timeout_seconds, + verify=verify_tls, + ) + + async def close(self) -> None: + await self.client.aclose() + + async def send_message(self, body: str) -> MatrixSendResult: + payload = { + "msgtype": self.message_type, + "body": body, + } + return await self._send_room_message(payload) + + async def send_image( + self, + file_path: str, + body: str | None = None, + ) -> MatrixSendResult: + path = Path(file_path) + if not path.exists() or not path.is_file(): + return MatrixSendResult( + ok=False, + status_code=None, + event_id=None, + error=f"Image file not found: {file_path}", + content_uri=None, + ) + + mime_type, _ = mimetypes.guess_type(path.name) + mime_type = mime_type or "image/png" + data = path.read_bytes() + + upload_result = await self._upload_bytes( + filename=path.name, + data=data, + content_type=mime_type, + ) + if not upload_result.ok: + return MatrixSendResult( + ok=False, + status_code=upload_result.status_code, + event_id=None, + error=upload_result.error, + content_uri=None, + ) + + payload = { + "msgtype": "m.image", + "body": body or path.name, + "url": upload_result.content_uri, + "info": { + "mimetype": mime_type, + "size": len(data), + }, + } + + send_result = await self._send_room_message(payload) + send_result.content_uri = upload_result.content_uri + return send_result + + async def _send_room_message(self, payload: dict) -> MatrixSendResult: + txn_id = str(uuid.uuid4()) + room_id_encoded = quote(self.room_id, safe="") + url = ( + f"{self.homeserver_url}" + f"/_matrix/client/v3/rooms/{room_id_encoded}" + f"/send/m.room.message/{txn_id}" + ) + + try: + response = await self._request_with_refresh( + method="PUT", + url=url, + json=payload, + headers={"Content-Type": "application/json"}, + ) + data = response.json() if response.content else {} + + if response.is_success: + return MatrixSendResult( + ok=True, + status_code=response.status_code, + event_id=data.get("event_id"), + error=None, + content_uri=None, + ) + + return MatrixSendResult( + ok=False, + status_code=response.status_code, + event_id=None, + error=str(data), + content_uri=None, + ) + except Exception as exc: + return MatrixSendResult( + ok=False, + status_code=None, + event_id=None, + error=str(exc), + content_uri=None, + ) + + async def _upload_bytes( + self, + filename: str, + data: bytes, + content_type: str, + ) -> MatrixUploadResult: + url = f"{self.homeserver_url}/_matrix/media/v3/upload" + + try: + response = await self._request_with_refresh( + method="POST", + url=url, + params={"filename": filename}, + content=data, + headers={"Content-Type": content_type}, + ) + payload = response.json() if response.content else {} + + if response.is_success: + return MatrixUploadResult( + ok=True, + status_code=response.status_code, + content_uri=payload.get("content_uri"), + error=None, + ) + + return MatrixUploadResult( + ok=False, + status_code=response.status_code, + content_uri=None, + error=str(payload), + ) + except Exception as exc: + return MatrixUploadResult( + ok=False, + status_code=None, + content_uri=None, + error=str(exc), + ) + + async def _request_with_refresh( + self, + method: str, + url: str, + **kwargs, + ) -> httpx.Response: + response = await self._request_once(method, url, **kwargs) + + if response.status_code == 401: + await self.token_manager.refresh_if_needed(force=True) + response = await self._request_once(method, url, **kwargs) + + return response + + async def _request_once( + self, + method: str, + url: str, + **kwargs, + ) -> httpx.Response: + token = await self.token_manager.get_access_token() + + headers = dict(kwargs.pop("headers", {}) or {}) + headers["Authorization"] = f"Bearer {token}" + + return await self.client.request( + method=method, + url=url, + headers=headers, + **kwargs, + ) \ No newline at end of file diff --git a/alert-processor/app/matrix_token_manager.py b/alert-processor/app/matrix_token_manager.py new file mode 100644 index 0000000..6cf8eeb --- /dev/null +++ b/alert-processor/app/matrix_token_manager.py @@ -0,0 +1,168 @@ +from __future__ import annotations + +import asyncio +import json +import logging +import time +from dataclasses import asdict, dataclass +from pathlib import Path + +import httpx + +logger = logging.getLogger(__name__) + + +@dataclass +class MatrixTokenState: + access_token: str + refresh_token: str + expires_at: float | None = None + + +class MatrixTokenManager: + def __init__( + self, + token_endpoint: str, + client_id: str, + client_secret: str | None, + initial_access_token: str, + initial_refresh_token: str, + initial_expires_in_seconds: int | None, + refresh_margin_seconds: int, + state_file: str, + timeout_seconds: float = 10, + verify_tls: bool = True, + ) -> None: + self.token_endpoint = token_endpoint + self.client_id = client_id + self.client_secret = client_secret or None + self.refresh_margin_seconds = refresh_margin_seconds + self.state_path = Path(state_file) + self._lock = asyncio.Lock() + self._stop_event = asyncio.Event() + self._refresh_task: asyncio.Task | None = None + self._refresh_failures = 0 + + self.client = httpx.AsyncClient( + timeout=timeout_seconds, + verify=verify_tls, + ) + + state = self._load_state() + if state is None: + expires_at = None + if initial_expires_in_seconds and initial_expires_in_seconds > 0: + expires_at = time.time() + initial_expires_in_seconds + + state = MatrixTokenState( + access_token=initial_access_token, + refresh_token=initial_refresh_token, + expires_at=expires_at, + ) + self._save_state(state) + + self._state = state + + def _load_state(self) -> MatrixTokenState | None: + if not self.state_path.exists(): + return None + + raw = json.loads(self.state_path.read_text(encoding="utf-8")) + return MatrixTokenState( + access_token=raw["access_token"], + refresh_token=raw["refresh_token"], + expires_at=raw.get("expires_at"), + ) + + def _save_state(self, state: MatrixTokenState) -> None: + self.state_path.write_text( + json.dumps(asdict(state), ensure_ascii=False, indent=2), + encoding="utf-8", + ) + + def _needs_refresh(self) -> bool: + if not self._state.refresh_token: + return False + if self._state.expires_at is None: + return False + return time.time() >= (self._state.expires_at - self.refresh_margin_seconds) + + async def get_access_token(self) -> str: + await self.refresh_if_needed() + return self._state.access_token + + async def refresh_if_needed(self, force: bool = False) -> MatrixTokenState: + async with self._lock: + if not force and not self._needs_refresh(): + return self._state + + data = { + "grant_type": "refresh_token", + "refresh_token": self._state.refresh_token, + "client_id": self.client_id, + } + if self.client_secret: + data["client_secret"] = self.client_secret + + response = await self.client.post( + self.token_endpoint, + data=data, + headers={"Content-Type": "application/x-www-form-urlencoded"}, + ) + response.raise_for_status() + payload = response.json() + + access_token = payload["access_token"] + refresh_token = payload.get("refresh_token", self._state.refresh_token) + expires_in = payload.get("expires_in") + expires_at = None + if expires_in is not None: + expires_at = time.time() + int(expires_in) + + self._state = MatrixTokenState( + access_token=access_token, + refresh_token=refresh_token, + expires_at=expires_at, + ) + self._save_state(self._state) + self._refresh_failures = 0 + + logger.info("Matrix OAuth token refreshed successfully") + return self._state + + def start_background_refresh(self) -> None: + if self._refresh_task is None: + self._refresh_task = asyncio.create_task(self._refresh_loop()) + + async def _refresh_loop(self) -> None: + while not self._stop_event.is_set(): + try: + await self.refresh_if_needed() + except Exception as exc: + self._refresh_failures += 1 + logger.exception("Matrix token background refresh failed: %s", exc) + + if self._refresh_failures > 0: + sleep_for = min(300, 30 * self._refresh_failures) + else: + sleep_for = 30 + if self._state.expires_at is not None: + remaining = int(self._state.expires_at - time.time() - self.refresh_margin_seconds) + sleep_for = max(5, min(60, remaining if remaining > 0 else 5)) + + try: + await asyncio.wait_for(self._stop_event.wait(), timeout=sleep_for) + except asyncio.TimeoutError: + pass + + async def close(self) -> None: + self._stop_event.set() + + if self._refresh_task is not None: + self._refresh_task.cancel() + try: + await self._refresh_task + except asyncio.CancelledError: + pass + + await self.client.aclose() \ No newline at end of file diff --git a/alert-processor/app/models.py b/alert-processor/app/models.py new file mode 100644 index 0000000..e797b31 --- /dev/null +++ b/alert-processor/app/models.py @@ -0,0 +1,101 @@ +from __future__ import annotations + +from datetime import datetime +from typing import Any + +from pydantic import BaseModel, Field + + +class NormalizedEvent(BaseModel): + source: str = "zabbix" + correlation_id: str + received_at: datetime + remote_addr: str | None = None + + event_id: str | None = None + problem_id: str | None = None + event_type: str = "problem" + timestamp: datetime | None = None + + severity: str | None = None + severity_code: int | None = None + + host: str | None = None + host_id: str | None = None + service: str | None = None + + trigger_name: str | None = None + trigger_id: str | None = None + item_id: str | None = None + + value: str | None = None + opdata: str | None = None + zabbix_url: str | None = None + + event_url: str | None = None + graph_url: str | None = None + graph_image_path: str | None = None + zabbix_context: dict[str, Any] = Field(default_factory=dict) + + tags: dict[str, str] = Field(default_factory=dict) + raw_payload: dict[str, Any] + + +class ProcessorForwardEnvelope(BaseModel): + event: NormalizedEvent + + +class NotificationDecision(BaseModel): + notify: bool + severity: str | None = None + channels: list[str] = Field(default_factory=list) + reason: str + routing_class: str + fingerprint: str | None = None + repeat_count: int = 1 + suppressed: bool = False + suppress_reason: str | None = None + event_phase: str = "problem" + open_incident_found: bool = False + recovered_from_severity: str | None = None + flap_detected: bool = False + flap_event_count: int = 0 + flap_reason: str | None = None + + triage_applied: bool = False + triage_source: str | None = None + triage_verdict: str | None = None + triage_reason: str | None = None + triage_classification: str | None = None + + correlation_applied: bool = False + correlation_role: str | None = None + correlation_group_id: str | None = None + correlation_kind: str | None = None + correlation_reason: str | None = None + correlation_source: str | None = None + correlation_confidence: str | None = None + parent_event_id: str | None = None + parent_correlation_id: str | None = None + root_cause_candidate: bool = False + correlated_event_count: int = 0 + + llm_enriched: bool = False + remediation_summary: str | None = None + remediation_steps: list[str] = Field(default_factory=list) + remediation_commands: list[str] = Field(default_factory=list) + + +class ProcessorAck(BaseModel): + accepted: bool + correlation_id: str + decision: NotificationDecision + message: str + + +class IngestAck(BaseModel): + accepted: bool + correlation_id: str + queued: bool + job_id: str + message: str \ No newline at end of file diff --git a/alert-processor/app/policy.py b/alert-processor/app/policy.py new file mode 100644 index 0000000..36615e2 --- /dev/null +++ b/alert-processor/app/policy.py @@ -0,0 +1,269 @@ +from __future__ import annotations + +from app.models import NormalizedEvent, NotificationDecision + + +HIGH_SEVERITIES = {"high", "disaster"} +AVERAGE_SEVERITIES = {"average"} +LOW_SEVERITIES = {"warning", "information", "not classified", "not_classified", "info"} + +RECOVERY_EVENT_TYPES = {"recovery", "resolved", "resolve", "ok", "clear", "closed"} + + +def normalize_severity(value: str | None) -> str | None: + if value is None: + return None + return value.strip().lower() + + +def normalize_event_phase(event_type: str | None, value: str | None = None) -> str: + raw = (event_type or "").strip().lower() + if raw in RECOVERY_EVENT_TYPES: + return "recovery" + + if not raw and str(value or "").strip() == "0": + return "recovery" + + return "problem" + + +def evaluate_event( + event: NormalizedEvent, + fingerprint: str, + repeat_count: int, +) -> NotificationDecision: + severity = normalize_severity(event.severity) + + if severity in HIGH_SEVERITIES: + return NotificationDecision( + notify=True, + severity=event.severity, + channels=["matrix", "mail"], + reason="Severity is High/Disaster: mandatory notification", + routing_class="high_priority", + fingerprint=fingerprint, + repeat_count=repeat_count, + event_phase="problem", + ) + + if severity in AVERAGE_SEVERITIES: + return NotificationDecision( + notify=True, + severity=event.severity, + channels=["matrix"], + reason="Severity is Average: notify by deterministic baseline policy", + routing_class="average_priority", + fingerprint=fingerprint, + repeat_count=repeat_count, + event_phase="problem", + ) + + if severity in LOW_SEVERITIES: + return NotificationDecision( + notify=False, + severity=event.severity, + channels=[], + reason="Severity is Warning or lower: held for triage/suppress pipeline", + routing_class="low_priority", + fingerprint=fingerprint, + repeat_count=repeat_count, + event_phase="problem", + ) + + return NotificationDecision( + notify=False, + severity=event.severity, + channels=[], + reason="Unknown severity: conservative hold until policy is expanded", + routing_class="unknown_priority", + fingerprint=fingerprint, + repeat_count=repeat_count, + event_phase="problem", + ) + + +def build_recovery_decision( + fingerprint: str, + repeat_count: int, + open_incident_found: bool, + previous_severity: str | None, + previous_channels: list[str] | None, + previous_routing_class: str | None, +) -> NotificationDecision: + severity_norm = normalize_severity(previous_severity) + + if not open_incident_found: + return NotificationDecision( + notify=False, + severity=previous_severity, + channels=[], + reason="Recovery received but no matching open incident was found", + routing_class="recovery_ignored", + fingerprint=fingerprint, + repeat_count=repeat_count, + event_phase="recovery", + open_incident_found=False, + recovered_from_severity=previous_severity, + ) + + if previous_routing_class == "triage_low_priority_notify": + return NotificationDecision( + notify=True, + severity=previous_severity, + channels=previous_channels or ["matrix"], + reason="Recovery matched a triaged low-severity incident", + routing_class="recovery_triage_low_priority", + fingerprint=fingerprint, + repeat_count=repeat_count, + event_phase="recovery", + open_incident_found=True, + recovered_from_severity=previous_severity, + ) + + if severity_norm in HIGH_SEVERITIES: + return NotificationDecision( + notify=True, + severity=previous_severity, + channels=previous_channels or ["matrix", "mail"], + reason="Recovery matched an open High/Disaster incident", + routing_class="recovery_high_priority", + fingerprint=fingerprint, + repeat_count=repeat_count, + event_phase="recovery", + open_incident_found=True, + recovered_from_severity=previous_severity, + ) + + if severity_norm in AVERAGE_SEVERITIES: + return NotificationDecision( + notify=True, + severity=previous_severity, + channels=previous_channels or ["matrix"], + reason="Recovery matched an open Average incident", + routing_class="recovery_average_priority", + fingerprint=fingerprint, + repeat_count=repeat_count, + event_phase="recovery", + open_incident_found=True, + recovered_from_severity=previous_severity, + ) + + return NotificationDecision( + notify=False, + severity=previous_severity, + channels=[], + reason="Recovery matched a low-priority incident: no notification by baseline policy", + routing_class="recovery_low_priority", + fingerprint=fingerprint, + repeat_count=repeat_count, + event_phase="recovery", + open_incident_found=True, + recovered_from_severity=previous_severity, + ) + + +def decision_supports_suppress(decision: NotificationDecision) -> bool: + return decision.routing_class == "average_priority" and decision.notify + + +def decision_supports_flap_suppress(decision: NotificationDecision) -> bool: + return decision.routing_class in { + "average_priority", + "recovery_average_priority", + } and decision.notify + + +def is_low_severity_problem_candidate( + decision: NotificationDecision, +) -> bool: + return decision.event_phase == "problem" and decision.routing_class == "low_priority" + + +def apply_suppress_window( + decision: NotificationDecision, + ttl_seconds: int, +) -> NotificationDecision: + return decision.model_copy( + update={ + "notify": False, + "channels": [], + "suppressed": True, + "suppress_reason": f"Suppress window active, {ttl_seconds}s remaining", + "reason": "Duplicate Average event suppressed by re-notify window", + } + ) + + +def apply_flap_suppress( + decision: NotificationDecision, + event_count: int, + window_seconds: int, +) -> NotificationDecision: + return decision.model_copy( + update={ + "notify": False, + "channels": [], + "suppressed": True, + "flap_detected": True, + "flap_event_count": event_count, + "flap_reason": f"Flapping detected: {event_count} phase changes within {window_seconds}s", + "reason": "Average event suppressed because the series is flapping", + } + ) + + +def apply_low_severity_triage( + decision: NotificationDecision, + verdict: str, + classification: str | None, + reason: str | None, + source: str, +) -> NotificationDecision: + verdict_norm = (verdict or "hold").strip().lower() + classification_norm = (classification or "unknown").strip().lower() + reason_text = (reason or "").strip() or "LLM triage applied" + + common = { + "triage_applied": True, + "triage_source": source, + "triage_verdict": verdict_norm, + "triage_reason": reason_text, + "triage_classification": classification_norm, + } + + if verdict_norm == "notify": + return decision.model_copy( + update={ + **common, + "notify": True, + "suppressed": False, + "suppress_reason": None, + "channels": ["matrix"], + "routing_class": "triage_low_priority_notify", + "reason": f"LLM triage marked low-severity event as actionable: {reason_text}", + } + ) + + if verdict_norm == "suppress": + return decision.model_copy( + update={ + **common, + "notify": False, + "suppressed": True, + "channels": [], + "routing_class": "triage_low_priority_suppressed", + "reason": "LLM triage suppressed low-severity event", + "suppress_reason": reason_text, + } + ) + + return decision.model_copy( + update={ + **common, + "notify": False, + "suppressed": False, + "channels": [], + "routing_class": "triage_low_priority_hold", + "reason": f"LLM triage kept low-severity event on hold: {reason_text}", + } + ) \ No newline at end of file diff --git a/alert-processor/app/processor_service.py b/alert-processor/app/processor_service.py new file mode 100644 index 0000000..8990c70 --- /dev/null +++ b/alert-processor/app/processor_service.py @@ -0,0 +1,744 @@ +from __future__ import annotations + +import logging + +from app.audit_logger import AuditLogger +from app.config import settings +from app.correlation import ( + CorrelationAssessment, + CorrelationRegistry, + apply_correlation_to_decision, + assess_correlation, +) +from app.fingerprint import build_fingerprint +from app.llm_correlation import LLMCorrelationAdapter, LLMCorrelationResult +from app.llm_remediation import LLMRemediationAdapter +from app.llm_triage import LLMTriageAdapter +from app.models import NotificationDecision, ProcessorForwardEnvelope +from app.notifications.dispatcher import NotificationDispatcher +from app.policy import ( + apply_flap_suppress, + apply_low_severity_triage, + apply_suppress_window, + build_recovery_decision, + decision_supports_flap_suppress, + decision_supports_suppress, + evaluate_event, + is_low_severity_problem_candidate, + normalize_event_phase, +) +from app.redis_repo import FlapState, OpenIncidentState, RedisStateRepository +from app.zabbix_enricher import ZabbixEnricher + +logger = logging.getLogger(__name__) + + +def _confidence_rank(value: str | None) -> int: + normalized = (value or "").strip().lower() + if normalized == "high": + return 3 + if normalized == "medium": + return 2 + if normalized == "low": + return 1 + return 0 + + +def _is_high_or_disaster(severity: str | None) -> bool: + value = (severity or "").strip().lower() + return value in {"high", "disaster"} + + +class ProcessorService: + def __init__( + self, + redis_repo: RedisStateRepository, + notification_dispatcher: NotificationDispatcher, + zabbix_enricher: ZabbixEnricher | None = None, + llm_remediation_adapter: LLMRemediationAdapter | None = None, + llm_triage_adapter: LLMTriageAdapter | None = None, + correlation_registry: CorrelationRegistry | None = None, + audit_logger: AuditLogger | None = None, + llm_correlation_adapter: LLMCorrelationAdapter | None = None, + ) -> None: + self.redis_repo = redis_repo + self.notification_dispatcher = notification_dispatcher + self.zabbix_enricher = zabbix_enricher + self.llm_remediation_adapter = llm_remediation_adapter + self.llm_triage_adapter = llm_triage_adapter + self.correlation_registry = correlation_registry + self.audit_logger = audit_logger + self.llm_correlation_adapter = llm_correlation_adapter + + async def process(self, envelope: ProcessorForwardEnvelope) -> NotificationDecision: + event = envelope.event + fingerprint = build_fingerprint(event) + repeat_count = 1 + event_phase = normalize_event_phase(event.event_type, event.value) + + async def audit(stage: str, status: str = "ok", details: dict | None = None) -> None: + if self.audit_logger is not None: + await self.audit_logger.log_stage( + correlation_id=event.correlation_id, + event_id=event.event_id, + stage=stage, + status=status, + details=details, + ) + + flap_state = FlapState( + active=False, + event_count=0, + phases=[], + window_seconds=settings.flap_window_seconds, + ) + + fp_state = await self.redis_repo.update_fingerprint_state( + fingerprint=fingerprint, + event=event, + ) + repeat_count = fp_state.count + + await self.redis_repo.save_event_snapshot( + event=event, + fingerprint=fingerprint, + repeat_count=repeat_count, + ) + + await audit( + "state_updated", + details={ + "fingerprint": fingerprint, + "repeat_count": repeat_count, + "event_phase": event_phase, + }, + ) + + if settings.flap_enabled: + flap_state = await self.redis_repo.record_phase_transition( + fingerprint=fingerprint, + event_phase=event_phase, + correlation_id=event.correlation_id, + event_id=event.event_id, + ) + await audit( + "flap_evaluated", + details={ + "flap_active": flap_state.active, + "flap_event_count": flap_state.event_count, + "window_seconds": flap_state.window_seconds, + }, + ) + + baseline_low_severity_candidate = False + + if event_phase == "recovery": + open_incident: OpenIncidentState | None = await self.redis_repo.get_open_incident( + fingerprint + ) + + decision = build_recovery_decision( + fingerprint=fingerprint, + repeat_count=repeat_count, + open_incident_found=open_incident is not None, + previous_severity=open_incident.severity if open_incident else None, + previous_channels=open_incident.channels if open_incident else None, + previous_routing_class=open_incident.routing_class if open_incident else None, + ) + + if ( + flap_state.active + and settings.flap_enabled + and settings.flap_apply_to_average + and decision_supports_flap_suppress(decision) + ): + decision = apply_flap_suppress( + decision=decision, + event_count=flap_state.event_count, + window_seconds=flap_state.window_seconds, + ) + + await self.redis_repo.clear_suppress_window(fingerprint) + await self.redis_repo.clear_open_incident(fingerprint) + + else: + decision = evaluate_event( + event=event, + fingerprint=fingerprint, + repeat_count=repeat_count, + ) + + if ( + flap_state.active + and settings.flap_enabled + and settings.flap_apply_to_average + and decision_supports_flap_suppress(decision) + ): + decision = apply_flap_suppress( + decision=decision, + event_count=flap_state.event_count, + window_seconds=flap_state.window_seconds, + ) + elif ( + settings.suppress_enabled + and settings.suppress_apply_to_average + and decision_supports_suppress(decision) + ): + suppress_state = await self.redis_repo.get_suppress_state(fingerprint) + + if suppress_state.active: + decision = apply_suppress_window( + decision=decision, + ttl_seconds=suppress_state.ttl_seconds, + ) + elif decision.notify: + await self.redis_repo.activate_suppress_window(fingerprint) + + baseline_low_severity_candidate = ( + settings.llm_triage_enabled + and self.llm_triage_adapter is not None + and is_low_severity_problem_candidate(decision) + and not decision.suppressed + ) + + await audit( + "baseline_policy_applied", + details={ + "notify": decision.notify, + "suppressed": decision.suppressed, + "routing_class": decision.routing_class, + "reason": decision.reason, + }, + ) + + should_enrich = self.zabbix_enricher is not None + if settings.zabbix_enrich_only_notify: + should_enrich = should_enrich and ( + (decision.notify and not decision.suppressed) or baseline_low_severity_candidate + ) + + if should_enrich and self.zabbix_enricher is not None: + try: + await self.zabbix_enricher.enrich_event(event) + await audit( + "zabbix_enrichment", + details={ + "event_url": event.event_url, + "graph_url": event.graph_url, + "graph_image_path": event.graph_image_path, + "zabbix_context_keys": sorted(list((event.zabbix_context or {}).keys())), + }, + ) + except Exception as exc: + await audit( + "zabbix_enrichment", + status="error", + details={"error": str(exc)}, + ) + logger.exception( + "Zabbix enrichment failed: correlation_id=%s event_id=%s error=%s", + event.correlation_id, + event.event_id, + exc, + ) + + if baseline_low_severity_candidate and self.llm_triage_adapter is not None: + triage_applied = False + + cached = await self.redis_repo.get_triage_cache( + fingerprint=fingerprint, + severity=decision.severity, + ) + if cached is not None: + decision = apply_low_severity_triage( + decision=decision, + verdict=cached.verdict, + classification=cached.classification, + reason=cached.reason, + source="cache", + ) + triage_applied = True + + await audit( + "llm_triage", + details={ + "source": "cache", + "verdict": cached.verdict, + "classification": cached.classification, + "reason": cached.reason, + }, + ) + + logger.info( + "LLM triage reused from cache: correlation_id=%s event_id=%s verdict=%s classification=%s ttl=%s", + event.correlation_id, + event.event_id, + cached.verdict, + cached.classification, + cached.ttl_seconds, + ) + + if not triage_applied: + try: + triage = await self.llm_triage_adapter.generate(envelope, decision) + if triage.ok: + decision = apply_low_severity_triage( + decision=decision, + verdict=triage.verdict, + classification=triage.classification, + reason=triage.reason, + source="llm", + ) + await self.redis_repo.save_triage_cache( + fingerprint=fingerprint, + severity=decision.severity, + verdict=triage.verdict, + classification=triage.classification, + reason=triage.reason, + source="llm", + ) + + await audit( + "llm_triage", + details={ + "source": "llm", + "verdict": triage.verdict, + "classification": triage.classification, + "reason": triage.reason, + }, + ) + else: + await audit( + "llm_triage", + status="error", + details={"error": triage.error}, + ) + logger.warning( + "LLM triage returned no usable content: correlation_id=%s event_id=%s error=%s", + event.correlation_id, + event.event_id, + triage.error, + ) + except Exception as exc: + await audit( + "llm_triage", + status="error", + details={"error": str(exc)}, + ) + logger.exception( + "LLM triage failed: correlation_id=%s event_id=%s error=%s", + event.correlation_id, + event.event_id, + exc, + ) + + if ( + settings.correlation_enabled + and self.correlation_registry is not None + and decision.event_phase == "problem" + ): + try: + zbx_context = event.zabbix_context or {} + scope_keys = list(zbx_context.get("correlation_scopes") or []) + + recent = await self.redis_repo.get_recent_correlation_events( + host=event.host, + scope_keys=scope_keys, + window_seconds=settings.correlation_window_seconds, + ) + + deterministic_assessment = assess_correlation( + envelope=envelope, + decision=decision, + recent_events=recent, + registry=self.correlation_registry, + ) + + chosen_assessment: CorrelationAssessment | None = ( + deterministic_assessment if deterministic_assessment.applied else None + ) + + strong_deterministic = ( + deterministic_assessment.applied + and deterministic_assessment.role in {"root", "child"} + ) + + if ( + self.llm_correlation_adapter is not None + and settings.llm_correlation_enabled + and not strong_deterministic + ): + try: + llm_result = await self.llm_correlation_adapter.generate( + envelope=envelope, + decision=decision, + recent_events=recent, + ) + + llm_assessment = self._build_llm_correlation_assessment( + llm_result=llm_result, + decision=decision, + event=envelope.event, + ) + + if llm_assessment is not None: + chosen_assessment = llm_assessment + await audit( + "llm_correlation", + details={ + "role": llm_assessment.role, + "kind": llm_assessment.kind, + "reason": llm_assessment.reason, + "confidence": llm_assessment.confidence, + "parent_event_id": llm_assessment.parent_event_id, + "parent_correlation_id": llm_assessment.parent_correlation_id, + "suppress_child": llm_assessment.suppress_child, + "scope_keys": scope_keys, + }, + ) + else: + await audit( + "llm_correlation", + status="skipped", + details={ + "reason": llm_result.reason, + "confidence": llm_result.confidence, + "error": llm_result.error, + "scope_keys": scope_keys, + }, + ) + except Exception as exc: + await audit( + "llm_correlation", + status="error", + details={"error": str(exc), "scope_keys": scope_keys}, + ) + logger.exception( + "LLM correlation fallback failed: correlation_id=%s event_id=%s error=%s", + event.correlation_id, + event.event_id, + exc, + ) + + if chosen_assessment is not None: + decision = apply_correlation_to_decision( + decision=decision, + assessment=chosen_assessment, + ) + + await self.redis_repo.save_correlation_event( + host=event.host or "", + event_id=event.event_id, + correlation_id=event.correlation_id, + kind=chosen_assessment.kind, + severity=decision.severity, + routing_class=decision.routing_class, + fingerprint=decision.fingerprint, + root_candidate=chosen_assessment.root_cause_candidate, + role=chosen_assessment.role, + group_id=chosen_assessment.group_id, + parent_event_id=chosen_assessment.parent_event_id, + parent_correlation_id=chosen_assessment.parent_correlation_id, + scope_keys=scope_keys, + tags=event.tags, + service=zbx_context.get("service") or event.service, + scope=zbx_context.get("scope"), + component=zbx_context.get("component"), + domain=zbx_context.get("domain"), + ) + + await audit( + "correlation", + details={ + "applied": chosen_assessment.applied, + "role": chosen_assessment.role, + "kind": chosen_assessment.kind, + "group_id": chosen_assessment.group_id, + "parent_event_id": chosen_assessment.parent_event_id, + "parent_correlation_id": chosen_assessment.parent_correlation_id, + "root_cause_candidate": chosen_assessment.root_cause_candidate, + "correlated_event_count": chosen_assessment.correlated_event_count, + "reason": chosen_assessment.reason, + "suppress_child": chosen_assessment.suppress_child, + "source": chosen_assessment.source, + "confidence": chosen_assessment.confidence, + "scope_keys": scope_keys, + }, + ) + + except Exception as exc: + await audit( + "correlation", + status="error", + details={"error": str(exc)}, + ) + logger.exception( + "Correlation baseline failed: correlation_id=%s event_id=%s error=%s", + event.correlation_id, + event.event_id, + exc, + ) + + if ( + decision.event_phase == "problem" + and decision.notify + and not decision.suppressed + and decision.routing_class in { + "high_priority", + "average_priority", + "triage_low_priority_notify", + } + ): + await self.redis_repo.upsert_open_incident( + fingerprint=fingerprint, + event=event, + decision=decision, + ) + + if ( + self.llm_remediation_adapter is not None + and decision.event_phase == "problem" + and decision.notify + and not decision.suppressed + ): + try: + remediation = await self.llm_remediation_adapter.generate(envelope, decision) + if remediation.ok: + decision = decision.model_copy( + update={ + "llm_enriched": True, + "remediation_summary": remediation.summary, + "remediation_steps": remediation.steps, + "remediation_commands": remediation.commands, + } + ) + await audit( + "llm_remediation", + details={ + "summary": remediation.summary, + "steps_count": len(remediation.steps), + "commands_count": len(remediation.commands), + }, + ) + else: + await audit( + "llm_remediation", + status="error", + details={"error": remediation.error}, + ) + logger.warning( + "LLM remediation returned no usable content: correlation_id=%s event_id=%s error=%s", + event.correlation_id, + event.event_id, + remediation.error, + ) + except Exception as exc: + await audit( + "llm_remediation", + status="error", + details={"error": str(exc)}, + ) + logger.exception( + "LLM remediation failed: correlation_id=%s event_id=%s error=%s", + event.correlation_id, + event.event_id, + exc, + ) + + if self.audit_logger is not None: + await self.audit_logger.log_decision( + envelope=envelope, + decision=decision, + ) + + logger.info( + "Processed event: correlation_id=%s event_id=%s event_phase=%s severity=%s " + "host=%s trigger=%s fingerprint=%s repeat_count=%s " + "open_incident_found=%s flap_detected=%s flap_event_count=%s " + "notify=%s suppressed=%s channels=%s reason=%s " + "triage_applied=%s triage_source=%s triage_verdict=%s triage_classification=%s triage_reason=%s " + "correlation_applied=%s correlation_role=%s correlation_kind=%s correlation_group_id=%s parent_event_id=%s root_cause_candidate=%s correlated_event_count=%s correlation_reason=%s correlation_source=%s correlation_confidence=%s " + "suppress_reason=%s flap_reason=%s recovered_from_severity=%s " + "event_url=%s graph_url=%s llm_enriched=%s", + event.correlation_id, + event.event_id, + decision.event_phase, + event.severity, + event.host, + event.trigger_name, + fingerprint, + repeat_count, + decision.open_incident_found, + decision.flap_detected, + decision.flap_event_count, + decision.notify, + decision.suppressed, + ",".join(decision.channels), + decision.reason, + decision.triage_applied, + decision.triage_source, + decision.triage_verdict, + decision.triage_classification, + decision.triage_reason, + decision.correlation_applied, + decision.correlation_role, + decision.correlation_kind, + decision.correlation_group_id, + decision.parent_event_id, + decision.root_cause_candidate, + decision.correlated_event_count, + decision.correlation_reason, + decision.correlation_source, + decision.correlation_confidence, + decision.suppress_reason, + decision.flap_reason, + decision.recovered_from_severity, + event.event_url, + event.graph_url, + decision.llm_enriched, + ) + + dispatch_report = await self.notification_dispatcher.dispatch( + envelope=envelope, + decision=decision, + ) + + delivery_payload = { + "attempted": dispatch_report.attempted, + "matrix_attempted": dispatch_report.matrix_attempted, + "matrix_sent": dispatch_report.matrix_sent, + "matrix_event_id": dispatch_report.matrix_event_id, + "matrix_error": dispatch_report.matrix_error, + "matrix_image_attempted": dispatch_report.matrix_image_attempted, + "matrix_image_sent": dispatch_report.matrix_image_sent, + "matrix_image_event_id": dispatch_report.matrix_image_event_id, + "matrix_image_mxc_uri": dispatch_report.matrix_image_mxc_uri, + "matrix_image_error": dispatch_report.matrix_image_error, + "mail_attempted": dispatch_report.mail_attempted, + "mail_sent": dispatch_report.mail_sent, + "mail_error": dispatch_report.mail_error, + "errors": dispatch_report.errors, + } + + if self.audit_logger is not None: + await self.audit_logger.log_delivery( + envelope=envelope, + delivery_payload=delivery_payload, + ) + + if dispatch_report.matrix_attempted: + if dispatch_report.matrix_sent: + logger.info( + "Matrix notification sent: correlation_id=%s event_id=%s matrix_event_id=%s", + event.correlation_id, + event.event_id, + dispatch_report.matrix_event_id, + ) + else: + logger.error( + "Matrix notification failed: correlation_id=%s event_id=%s error=%s", + event.correlation_id, + event.event_id, + dispatch_report.matrix_error, + ) + + if dispatch_report.matrix_image_attempted: + if dispatch_report.matrix_image_sent: + logger.info( + "Matrix graph image sent: correlation_id=%s event_id=%s matrix_event_id=%s content_uri=%s", + event.correlation_id, + event.event_id, + dispatch_report.matrix_image_event_id, + dispatch_report.matrix_image_mxc_uri, + ) + else: + logger.error( + "Matrix graph image failed: correlation_id=%s event_id=%s error=%s", + event.correlation_id, + event.event_id, + dispatch_report.matrix_image_error, + ) + + if dispatch_report.mail_attempted: + if dispatch_report.mail_sent: + logger.info( + "Mail notification sent: correlation_id=%s event_id=%s recipient=%s", + event.correlation_id, + event.event_id, + settings.mail_to, + ) + else: + logger.error( + "Mail notification failed: correlation_id=%s event_id=%s error=%s", + event.correlation_id, + event.event_id, + dispatch_report.mail_error, + ) + + return decision + + def _build_llm_correlation_assessment( + self, + llm_result: LLMCorrelationResult, + decision: NotificationDecision, + event, + ) -> CorrelationAssessment | None: + if not llm_result.ok: + return None + + min_confidence_rank = _confidence_rank(settings.llm_correlation_min_confidence) + actual_confidence_rank = _confidence_rank(llm_result.confidence) + + role = llm_result.role + confidence = llm_result.confidence or "low" + + if role in {"root", "child"} and actual_confidence_rank < min_confidence_rank: + return None + + if role == "child" and not (llm_result.parent_event_id or llm_result.parent_correlation_id): + return None + + suppress_child = False + if ( + role == "child" + and llm_result.suppress_child + and actual_confidence_rank >= _confidence_rank("high") + and not _is_high_or_disaster(decision.severity) + ): + suppress_child = True + + if role == "root": + group_id = event.event_id or event.correlation_id + return CorrelationAssessment( + applied=True, + role="root", + kind=llm_result.kind, + group_id=group_id, + reason=llm_result.reason, + parent_event_id=None, + parent_correlation_id=None, + root_cause_candidate=True, + correlated_event_count=0, + suppress_child=False, + source="llm", + confidence=confidence, + ) + + if role == "child": + group_id = llm_result.parent_event_id or llm_result.parent_correlation_id + return CorrelationAssessment( + applied=True, + role="child", + kind=llm_result.kind, + group_id=group_id, + reason=llm_result.reason, + parent_event_id=llm_result.parent_event_id, + parent_correlation_id=llm_result.parent_correlation_id, + root_cause_candidate=False, + correlated_event_count=0, + suppress_child=suppress_child, + source="llm", + confidence=confidence, + ) + + return None