"""Structured operational events with defensive redaction and retention.""" from __future__ import annotations import json import re from datetime import datetime, timedelta, timezone from sqlalchemy import delete, select from .config import settings from .models import AgentEvent EVENT_TYPES = { "AGENT_STARTED", "AGENT_STOPPED", "HEARTBEAT", "PREFLIGHT_PASS", "PREFLIGHT_FAIL", "JOB_CLAIMED", "JOB_STARTED", "JOB_COMPLETED", "JOB_FAILED", "PRODUCT_BLOCKED", "BLOCKER_RESOLVED", "DATASET_CHANGED", "COMPANY_CONTEXT_CHANGED", "SCHEMA_CHANGED", "WRITE_PLANNED", "WRITE_COMPLETED", "WRITE_FAILED", "CONNECTION_ERROR", "RETRY", "VERSION_CHANGED", "ADMIN_ACTION", } SEVERITIES = {"INFO", "WARNING", "ERROR", "CRITICAL"} SECRET_RE = re.compile(r"(?i)(password|fb_password|token|api[_-]?key|secret|authorization|connection[_-]?string)\s*[:=]\s*[^,;\s]+") def _redact(value): if isinstance(value, dict): return {str(key): "[REDACTED]" if str(key).lower() in {"password", "fb_password", "token", "api_key", "secret", "authorization", "connection_string"} else _redact(item) for key, item in value.items()} if isinstance(value, list): return [_redact(item) for item in value[:100]] if isinstance(value, str): return SECRET_RE.sub(lambda match: f"{match.group(1)}=[REDACTED]", value) return value def sanitize_event(message: str, context: dict | None = None) -> tuple[str, dict]: safe_message = _redact(str(message or ""))[: settings.web_max_event_message] safe_context = _redact(context or {}) encoded = json.dumps(safe_context, ensure_ascii=False, default=str) if len(encoded.encode()) > settings.web_max_event_context_bytes: encoded = encoded.encode()[: settings.web_max_event_context_bytes].decode("utf-8", errors="ignore") safe_context = {"truncated": True, "payload": encoded} return safe_message, safe_context def record_event(session, *, tenant_id: int, installation_id: int, agent_id: int, severity: str, event_type: str, message: str, context: dict | None = None, job_id: int | None = None, product_code: str | None = None, blocker_id: int | None = None, is_audit: bool = False, occurred_at: datetime | None = None): severity = severity.upper() event_type = event_type.upper() if severity not in SEVERITIES: raise ValueError("invalid event severity") if event_type not in EVENT_TYPES: raise ValueError("invalid event type") safe_message, safe_context = sanitize_event(message, context) event = AgentEvent(tenant_id=tenant_id, installation_id=installation_id, agent_id=agent_id, severity=severity, event_type=event_type, message=safe_message, structured_context=safe_context, job_id=job_id, product_code=product_code, blocker_id=blocker_id, is_audit=is_audit, occurred_at=occurred_at or datetime.now(timezone.utc)) session.add(event) return event def purge_expired_events(session, now: datetime | None = None, retention_days: int | None = None) -> int: cutoff = (now or datetime.now(timezone.utc)) - timedelta(days=retention_days if retention_days is not None else settings.agent_event_retention_days) result = session.execute(delete(AgentEvent).where(AgentEvent.occurred_at < cutoff, AgentEvent.is_audit.is_(False), AgentEvent.severity != "CRITICAL")) session.commit() return result.rowcount or 0 def list_events(session, installation_db_id: int, *, page: int = 1, page_size: int = 50, severity: str | None = None, event_type: str | None = None, job_id: int | None = None, product_code: str | None = None, since: datetime | None = None, until: datetime | None = None): page = max(1, page); page_size = min(100, max(1, page_size)) query = select(AgentEvent).where(AgentEvent.installation_id == installation_db_id) for column, value in ((AgentEvent.severity, severity), (AgentEvent.event_type, event_type), (AgentEvent.job_id, job_id), (AgentEvent.product_code, product_code)): if value is not None: query = query.where(column == value) if since: query = query.where(AgentEvent.occurred_at >= since) if until: query = query.where(AgentEvent.occurred_at <= until) query = query.order_by(AgentEvent.occurred_at.desc(), AgentEvent.id.desc()) rows = session.scalars(query.offset((page - 1) * page_size).limit(page_size)).all() return rows