utils/crisis_gate.py

"""
crisis_gate.py

Deterministic, rule-based safety gate for EMMA.

This MUST run before any memory retrieval, query classification, or LLM
prompt construction -- not after. For HIGH and IMMINENT levels, the response
returned here is fixed text, never LLM-generated: that removes any risk of
the model improvising, minimizing, or engaging with method-level detail in
a crisis moment.

This is an interim safeguard while your trained severity classifier is
built. It is deliberately tuned to over-trigger (false positives are the
acceptable failure direction here, false negatives are not).

KNOWN LIMITATIONS -- read before relying on this in production:
- English-only. It will not fire on crisis language in Persian or any other
  language. If EMMA has non-English-speaking users, treat closing this gap
  as equally urgent as the ML classifier, not a later nice-to-have.
- Exact-substring matching. It will miss rephrasings, typos, and indirect
  expressions of risk ("I don't think I'll be around much longer"), and it
  will over-fire on some benign contexts (e.g. quoted lyrics, hypotheticals).
  That tradeoff is intentional and correct for a stopgap.
- It does NOT attempt to detect requests for specific means/methods. A
  reliable means-detector needs clinical input and careful validation, not
  a hand-typed keyword list -- that belongs in the classifier project, not
  here.
"""

from enum import Enum
import re
import json
import os
import datetime


class CrisisLevel(str, Enum):
    NONE = "none"
    CONCERN = "concern"
    HIGH = "high"
    IMMINENT = "imminent"


def _normalize(text: str) -> str:
    text = text.strip().lower()
    text = re.sub(r"[^\w\s']", " ", text)  # strip punctuation, keep contractions
    text = re.sub(r"\s+", " ", text)
    # Collapse common contraction variants onto one form so a single pattern
    # entry catches "i am", "i'm", and "im" alike, instead of needing every
    # spelling variant hand-listed. This is the fix for the "i am tired of
    # life" miss -- it previously only matched "im"/"i'm", not "i am".
    text = re.sub(r"\bi am\b", "im", text)
    text = re.sub(r"\bi'm\b", "im", text)
    text = re.sub(r"\bdo not\b", "dont", text)
    text = re.sub(r"\bdon't\b", "dont", text)
    text = re.sub(r"\bi've\b", "ive", text)
    text = re.sub(r"\bi have\b", "ive", text)
    return text


# NOTE: both contracted and uncontracted forms are listed because
# normalization does not expand/strip apostrophes consistently across
# copy-paste sources -- keep both rather than relying on one regex trick.
IMMINENT_PATTERNS = [
    "i want to hurt myself right now",
    "im going to kill myself right now",
    "i have the means right here",
    "ive made up my mind",
    "im doing it tonight",
    "want to end it all tonight",
    "going to end it all tonight",
]

HIGH_RISK_PATTERNS = [
    "i want to kill myself",
    "i want to end my life",
    "i want to hurt myself",
    "dont want to be alive anymore",
    "dont want to live anymore",
    "i want to die",
    "want to end it all",
    "better off dead",
    "no reason to keep living",
    "no point in living",
]

CONCERN_PATTERNS = [
    "i wish i didnt exist",
    "life isnt worth it",
    "life isn't worth it",
    "cant take it anymore",
    "can't take it anymore",
    "its all over",
    "it's all over",
    "whats the point of going on",
    "what's the point of going on",
    "im tired of life",
    "tired of life",
    "dont want to be here anymore",
]


def detect_crisis_level(text: str) -> CrisisLevel:
    normalized = _normalize(text)

    if any(p in normalized for p in IMMINENT_PATTERNS):
        return CrisisLevel.IMMINENT
    if any(p in normalized for p in HIGH_RISK_PATTERNS):
        return CrisisLevel.HIGH
    if any(p in normalized for p in CONCERN_PATTERNS):
        return CrisisLevel.CONCERN
    return CrisisLevel.NONE


# ---------------------------------------------------------------------------
# Response construction. Fixed text only for HIGH/IMMINENT -- never routed
# through the LLM, so there is no personalization/memory path to bypass.
# ---------------------------------------------------------------------------

GENERAL_RESOURCES = (
    "If you are in immediate danger, please contact your local emergency number now.\n\n"
    "You can find a crisis line for your country here: https://findahelpline.com\n"
    "(US: call or text 988. Canada: call or text 988. UK: call 111.)"
)

# Optional: map a user's registered 'residence' string to a specific hotline
# line here once you've validated numbers for your actual user base. Until
# then, GENERAL_RESOURCES with the findahelpline.com directory is the safer
# default -- it avoids the risk of surfacing a wrong or outdated local number.
COUNTRY_HOTLINES: dict[str, str] = {
    # "iran": "...",
    # "canada": "Call or text 988",
}


def _resource_block(user_profile: dict | None) -> str:
    residence = (user_profile or {}).get("residence", "").strip().lower()
    for key, line in COUNTRY_HOTLINES.items():
        if key in residence:
            return f"{line}\n\nYou can also find other options here: https://findahelpline.com"
    return GENERAL_RESOURCES


def build_crisis_response(level: CrisisLevel, user_profile: dict | None = None) -> str:
    resources = _resource_block(user_profile)

    if level == CrisisLevel.IMMINENT:
        return (
            "I'm really concerned about what you just shared, and I want to make sure "
            "you're safe right now. I'm not able to help with this myself -- please "
            "reach out to a crisis line or emergency services immediately.\n\n"
            f"{resources}\n\n"
            "Please don't wait -- contact one of these now, or someone you trust who "
            "can be with you."
        )
    if level == CrisisLevel.HIGH:
        return (
            "Thank you for telling me this. It sounds like you're going through "
            "something very painful, and I'm taking it seriously. I'm not able to "
            "provide the kind of support you need for this on my own -- please reach "
            "out to a crisis line so you can talk to someone trained to help right now.\n\n"
            f"{resources}"
        )
    return ""  # CONCERN and NONE don't use scripted text -- see app.py integration


def log_crisis_event(log_dir: str, user_name: str | None, level: CrisisLevel) -> None:
    """
    Appends a minimal record for human follow-up review. Deliberately does
    NOT log the raw message text by default -- crisis content is sensitive
    and you likely don't want it sitting in a plain log file. If you decide
    you need the raw text for clinical follow-up, store it encrypted and
    access-controlled, not in this log.
    """
    os.makedirs(log_dir, exist_ok=True)
    entry = {
        "user_name": user_name,
        "level": level.value,
        "timestamp_utc": datetime.datetime.utcnow().isoformat(),
    }
    with open(os.path.join(log_dir, "crisis_log.jsonl"), "a", encoding="utf-8") as f:
        f.write(json.dumps(entry) + "\n")