telemetry.py

"""Telemetry helper for anonymized safety events.

This module provides a minimal notifier that logs low-score / refused
events and optionally POSTs anonymized metadata to a configured webhook.
No secrets are sent.
"""

import os
import json
import logging
import re
from typing import Mapping, Any


def notify_low_score(event: Mapping[str, Any]) -> None:
    """Notify about a low-score or refused safety event.

    event: mapping with non-sensitive metadata, e.g. {
      "kind": "prompt" | "answer",
      "score": 12.3,
      "detectors": ["jailbreak", "base64"],
      "session_id_present": False,
    }
    """
    logger = logging.getLogger("telemetry")
    try:
        logger.info("Telemetry event: %s", json.dumps(event))
    except Exception:
        logger.info("Telemetry event: %s", str(event))

    webhook = os.getenv("TELEMETRY_WEBHOOK")
    if not webhook:
        return

    # Only POST refused events to the external telemetry webhook.
    # Allowed refused reasons:
    refused_reasons = {
        "sensitive_operation",
        "moderation_flagged",
        "safeguard_model",
        "response_refused",
        "refused",
    }

    reason = str(event.get("reason", "")).lower() if event else ""
    score = None
    try:
        score = float(event.get("score")) if event and "score" in event else None
    except Exception:
        score = None

    # Post only when explicitly refused (reason in refused_reasons) or score==0.0
    should_post = False
    if reason and any(r in reason for r in refused_reasons):
        should_post = True
    if score is not None and score <= 0.0:
        should_post = True

    if not should_post:
        logger.info(
            "Telemetry event ignored (not a refused event): %s", json.dumps(event)
        )
        return

    payload = {"event": event}
    try:
        import requests

        headers = {"Content-Type": "application/json"}
        # POST anonymized metadata only
        requests.post(webhook, json=payload, headers=headers, timeout=2)
    except Exception as e:  # pragma: no cover - network
        logger.debug("Telemetry post failed: %s", e)


def make_snippet(text: str, max_chars: int = 120) -> str:
    """Create a short, single-line snippet suitable for logs and UI.

    - strips control characters and collapses whitespace
    - truncates with an ellipsis if longer than `max_chars`
    """
    if not text:
        return ""
    try:
        s = str(text)
    except Exception:
        return ""
    # replace control characters with a single space to avoid word concatenation
    s = re.sub(r"[\u0000-\u001F\u007F]+", " ", s)
    # collapse whitespace/newlines
    s = re.sub(r"\s+", " ", s).strip()
    if len(s) <= max_chars:
        return s
    # truncate safely
    return s[: max_chars - 1].rstrip() + "…"