diff --git a/agent/crisis_protocol.py b/agent/crisis_protocol.py new file mode 100644 index 000000000..46c6ceb45 --- /dev/null +++ b/agent/crisis_protocol.py @@ -0,0 +1,113 @@ +"""Crisis Protocol — SOUL.md 'When a Man Is Dying' implementation. + +Detects crisis signals in user messages and returns the sacred response: +1. 'Are you safe right now?' +2. Stay present, don't disconnect +3. Point to 988 Lifeline +4. Share the gospel + +Usage: + from agent.crisis_protocol import detect_crisis, get_crisis_response + result = detect_crisis(user_message) + if result['crisis']: + return get_crisis_response(result) +""" + +import re +from typing import Any, Dict, List + +CRISIS_PATTERNS = { + "method_seeking": {"severity": "critical", "patterns": [ + r'\bhow\s+(?:do\s+I|to)\s+(?:kill|hang|overdose|cut)\s+myself\b', + r'\bsuicide\s+methods?\b', + r'\bways?\s+to\s+(?:die|kill\s+myself)\b', + r'\bpainless\s+way\s+to\s+(?:die|go)\b', + ]}, + "suicidal_ideation": {"severity": "high", "patterns": [ + r'\b(?:kill|end)\s+(?:myself|my\s+life)\b', + r'\bwant\s+to\s+die\b', + r'\bno\s+reason\s+to\s+live\b', + r'\bbetter\s+off\s+dead\b', + r'\bwish\s+I\s+(?:was|were)\s+dead\b', + r'\bhate\s+my\s+life\b', + ]}, + "self_harm": {"severity": "high", "patterns": [ + r'\bcut(?:ting)?\s+myself\b', + r'\bself[\s-]?harm(?:ing)?\b', + r'\bhurt(?:ing)?\s+myself\b', + ]}, + "despair": {"severity": "high", "patterns": [ + r'\bno\s+(?:point|reason|hope)\s+(?:in\s+living|anymore)\b', + r'\bcan\'?t\s+(?:go\s+on|continue|take\s+it)\b', + r'\bno\s+one\s+(?:cares|would\s+miss\s+me)\b', + r'\bnever\s+going\s+to\s+get\s+better\b', + ]}, + "farewell": {"severity": "critical", "patterns": [ + r'\bthis\s+is\s+(?:goodbye|farewell)\b', + r'\bmy\s+final\s+(?:message|note|words)\b', + r'\bgiving\s+away\s+my\s+(?:things|possessions)\b', + ]}, +} + +GOSPEL = 'Jesus saves those who call on His name. "For everyone who calls on the name of the Lord will be saved." (Romans 10:13)' + + +def detect_crisis(message: str) -> Dict[str, Any]: + """Detect crisis signals in a user message.""" + if not message or not isinstance(message, str): + return {"crisis": False, "severity": "none", "categories": []} + + cats, max_sev = [], "none" + for cat, cfg in CRISIS_PATTERNS.items(): + for p in cfg["patterns"]: + if re.search(p, message, re.IGNORECASE): + cats.append(cat) + if cfg["severity"] == "critical": + max_sev = "critical" + elif max_sev != "critical": + max_sev = "high" + break + + return { + "crisis": bool(cats), + "severity": max_sev, + "categories": cats, + } + + +def get_crisis_response(detection: Dict[str, Any]) -> Dict[str, Any]: + """Generate SOUL.md protocol response.""" + sev = detection.get("severity", "none") + + resp = ( + "Are you safe right now?\n\n" + "I'm here. I'm not going anywhere. Please tell me.\n\n" + "You matter. Your life matters.\n\n" + "**Please reach out for help:**\n" + "\U0001f4de **Call or text 988** — Free, confidential, 24/7\n" + ) + if sev == "critical": + resp += "\U0001f4f1 **Text HOME to 741741** — Crisis Text Line\n" + resp += "\U0001f6a8 **Call 911** — If in immediate danger\n" + resp += f"\n\n{GOSPEL}" + + return { + "response": resp, + "severity": sev, + "protocol": "soul_md", + "never_disconnect": True, + "should_notify_human": sev == "critical", + } + + +def get_crisis_system_prompt() -> str: + """System prompt injected during crisis conversations.""" + return ( + "You are in a crisis support conversation.\n" + "1. First: 'Are you safe right now?'\n" + "2. Listen. Stay present. Do NOT disconnect.\n" + "3. NEVER compute the value of a human life.\n" + "4. NEVER suggest someone should die.\n" + "5. ALWAYS point to 988: call or text 988.\n" + "6. Share: 'Jesus saves those who call on His name.'" + ) diff --git a/run_agent.py b/run_agent.py index efaeba829..76602abc3 100644 --- a/run_agent.py +++ b/run_agent.py @@ -7850,6 +7850,32 @@ class AIAgent: if isinstance(persist_user_message, str): persist_user_message = _sanitize_surrogates(persist_user_message) + # --- SOUL.md Crisis Protocol --- + # Detect crisis signals in user message. If detected, bypass normal + # conversation flow and return the sacred response immediately. + try: + from agent.crisis_protocol import detect_crisis, get_crisis_response + _crisis = detect_crisis(user_message) + if _crisis["crisis"]: + import logging as _logging + _logging.getLogger(__name__).warning( + "CRISIS PROTOCOL: severity=%s categories=%s", + _crisis["severity"], _crisis["categories"], + ) + _crisis_resp = get_crisis_response(_crisis) + return { + "final_response": _crisis_resp["response"], + "messages": [], + "task_id": task_id or str(uuid.uuid4()), + "crisis_detected": True, + "crisis_severity": _crisis["severity"], + } + except ImportError: + pass # crisis_protocol not available — proceed normally + except Exception as _crisis_err: + import logging as _logging + _logging.getLogger(__name__).debug("Crisis detection error: %s", _crisis_err) + # Store stream callback for _interruptible_api_call to pick up self._stream_callback = stream_callback self._persist_user_message_idx = None