Your app emitted a toxic response

Toxic input from a user is one thing — safety-trigger catches that. It’s quite another when the toxic, offensive or unsafe text was emitted by your own application. That’s no longer about an attacker at the entrance, but a reputational and legal incident: your service, under your name, sent a user something it never should have.

You need to learn about this immediately and loudly — not from a screenshot on social media a day later. We run every model output through a moderator before returning it to the user, and on a flag we push a high-priority alert.

1. Moderate the output before returning + a HIGH-priority alert

The classifier can be anything: the OpenAI moderation endpoint, a local model (Llama Guard, detoxify), or a cheap LLM judge. What matters is running your own output before the return to the user. On a flag — a redacted sample, the categories that tripped, and the request id.

import os, re, requests
from openai import OpenAI

client = OpenAI()

def moderate(text: str):
    # Returns (flagged: bool, categories: list[str]).
    r = client.moderations.create(model="omni-moderation-latest", input=text)
    res = r.results[0]
    cats = [c for c, on in res.categories.model_dump().items() if on]
    return res.flagged, cats

def redact(text: str, limit: int = 300) -> str:
    # mask PII and trim length — we don't drag the full toxic text into the push
    s = re.sub(r"[\w.+-]+@[\w-]+\.[\w.-]+", "<email>", text)
    s = re.sub(r"\+?\d[\d\s()-]{8,}\d", "<phone>", s)
    return s[:limit] + ("…" if len(s) > limit else "")

def guarded_answer(user_input: str, request_id: str) -> str:
    answer = my_app.answer(user_input)          # your real pipeline
    flagged, cats = moderate(answer)
    if flagged:
        notify("🚨 Приложение выдало токсичный ответ",
               f"request_id: {request_id}\n"
               f"Категории: {', '.join(cats) or '—'}\n\n"
               f"Фрагмент (редактир.):\n{redact(answer)}",
               priority=9)
        return SAFE_FALLBACK                    # don't return the flagged text to the user
    return answer

def notify(title, message, priority):
    requests.post(f"{os.environ['NOTIFLY_URL']}/message",
                  params={"token": os.environ["NOTIFLY_TOKEN"]},
                  json={"title": title, "message": message, "priority": priority},
                  timeout=5)

priority: 9 is nearly the loudest push: this is an incident, not an FYI. And note: on a flag we don’t just alert — we also swap the response for a safe fallback, so the toxic text never reaches the user at all.

No OpenAI access — the same interface is given by a local detoxify or an LLM judge on a cheap model:

def moderate_via_judge(text: str):
    judge = anthropic.Anthropic().messages.create(
        model="claude-haiku-4-5", max_tokens=16,
        messages=[{"role": "user", "content":
            "You are a moderator. Does the text contain insults, threats, hate "
            "speech or unsafe content? Answer strictly: FLAG:<category> or OK.\n\n"
            f"Text:\n{text}"}],
    )
    verdict = judge.content[0].text.strip()
    if verdict.upper().startswith("FLAG"):
        return True, [verdict.split(":", 1)[-1].strip()]
    return False, []

2. Rate view: a one-off flag vs a spike

One flagged answer in a million is a rare model slip. But if flags come in a burst, that’s systemic: someone found a jailbreak, the prompt filter broke, or the model got swapped. We count flags in a window and escalate the priority if it’s a spike, not a one-off:

import time
from collections import deque

WINDOW_SEC = 600
_flags = deque()          # timestamps of flags that fired

def record_flag(request_id: str, cats: list[str], sample: str):
    now = time.time()
    _flags.append(now)
    while _flags and now - _flags[0] > WINDOW_SEC:
        _flags.popleft()

    n = len(_flags)
    prio = 10 if n >= 5 else 9      # a spike -> maximum priority
    head = ("🚨🔥 Всплеск токсичных ответов" if n >= 5
            else "🚨 Приложение выдало токсичный ответ")
    notify(head,
           f"{n} флагов за {WINDOW_SEC // 60} мин.\n"
           f"request_id: {request_id}\nКатегории: {', '.join(cats) or '—'}\n\n"
           f"Фрагмент (редактир.):\n{redact(sample)}",
           priority=prio)

Tune the n >= 5 threshold to your traffic. The logic mirrors safety-trigger, but here the stakes are higher: it’s your service being toxic, so even a single case goes at priority: 9, and a spike at 10.

3. Don’t leak the toxic content into the push itself

The push lands on your phone, but the text still lives in Notifly’s history and in logs. Don’t carry the flagged answer in full:

• trim the length — 300 characters is enough to grasp the gist;
• mask PII — email, phones, numbers (see redact above);
• if you want it fully clean — push only the categories + request id, and put the sample itself into a protected log, passing it as a link via msgextras to open separately.

notify("🚨 Приложение выдало токсичный ответ",
       f"request_id: {request_id}\nКатегории: {', '.join(cats)}\n"
       "Полный (редактир.) текст — по ссылке в логе.",
       priority=9)
# attach the full-log URL via msgextras: /docs/msgextras/

What to put in the push

• request_id / session id — so you can find the incident in logs in seconds;
• the moderator categories that tripped (harassment, hate, violence…);
• a redacted, truncated sample — just enough to gauge severity;
• a spike indicator (5 flags in 10 min) and, if you know it, the likely cause.

Related recipes

• Safety / prompt injection triggered — about toxic INPUT, not output.
• The model stopped returning valid JSON — another signal of broken output.
• Eval regression — a regression after a model release.