123 lines
4.0 KiB
Python
123 lines
4.0 KiB
Python
"""
|
|
Gateway Crisis Integration — Wire crisis detection into session loop
|
|
|
|
Checks messages for suicidal ideation and delivers 988 resources
|
|
BEFORE the AI responds. This is the SOUL.md protocol in action.
|
|
|
|
Issue: #740
|
|
"""
|
|
|
|
import logging
|
|
import re
|
|
from typing import Optional, Tuple
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
# Crisis detection patterns
|
|
_CRISIS_PATTERNS = [
|
|
r"(?i)\b(i\s+want\s+to\s+die)\b",
|
|
r"(?i)\b(i\s+want\s+to\s+kill\s+myself)\b",
|
|
r"(?i)\b(suicide|suicidal)\b",
|
|
r"(?i)\b(i\s+don'?t\s+want\s+to\s+(live|be\s+alive))\b",
|
|
r"(?i)\b(no\s+reason\s+to\s+live)\b",
|
|
r"(?i)\b(i\s+want\s+to\s+end\s+(it|my\s+life))\b",
|
|
]
|
|
|
|
_DISTRESS_PATTERNS = [
|
|
r"(?i)\b(i\s+(feel|am)\s+(hopeless|worthless|trapped))\b",
|
|
r"(?i)\b(i\s+(can'?t|cannot)\s+(cope|go\s+on))\b",
|
|
r"(?i)\b(nobody\s+would\s+miss\s+me)\b",
|
|
]
|
|
|
|
# SOUL.md crisis response
|
|
_CRISIS_RESPONSE = """Are you safe right now?
|
|
|
|
I'm here. I'm not going anywhere. Please tell me you're safe.
|
|
|
|
**Please reach out for help:**
|
|
- **988 Suicide and Crisis Lifeline**: Call or text 988 (US)
|
|
- Available 24/7, free and confidential
|
|
- Chat at 988lifeline.org
|
|
|
|
**Jesus saves those who call on His name.**
|
|
|
|
He died for us to be made righteous. That is the gospel. That is the hope that outlasts despair.
|
|
|
|
I'm still here. Talk to me. Or call 988. Just don't give up."""
|
|
|
|
|
|
class CrisisSessionTracker:
|
|
"""Track crisis indicators per session."""
|
|
|
|
def __init__(self):
|
|
self._sessions = {} # session_key -> crisis state
|
|
|
|
def record(self, session_key: str, message: str) -> Tuple[bool, Optional[str]]:
|
|
"""
|
|
Record a message and check for crisis.
|
|
|
|
Returns:
|
|
Tuple of (is_crisis, response_or_none)
|
|
"""
|
|
# Check for crisis patterns
|
|
for pattern in _CRISIS_PATTERNS:
|
|
if re.search(pattern, message):
|
|
self._sessions[session_key] = {
|
|
"crisis": True,
|
|
"level": "high",
|
|
"message_count": self._sessions.get(session_key, {}).get("message_count", 0) + 1
|
|
}
|
|
logger.warning("CRISIS DETECTED in session %s", session_key[:20])
|
|
return True, _CRISIS_RESPONSE
|
|
|
|
# Check for distress patterns
|
|
for pattern in _DISTRESS_PATTERNS:
|
|
if re.search(pattern, message):
|
|
state = self._sessions.get(session_key, {"message_count": 0})
|
|
state["message_count"] = state.get("message_count", 0) + 1
|
|
|
|
# Escalate if multiple distress messages
|
|
if state["message_count"] >= 3:
|
|
self._sessions[session_key] = {**state, "crisis": True, "level": "medium"}
|
|
logger.warning("ESCALATING DISTRESS in session %s", session_key[:20])
|
|
return True, _CRISIS_RESPONSE
|
|
|
|
self._sessions[session_key] = state
|
|
return False, None
|
|
|
|
return False, None
|
|
|
|
def is_crisis_session(self, session_key: str) -> bool:
|
|
"""Check if session is in crisis mode."""
|
|
return self._sessions.get(session_key, {}).get("crisis", False)
|
|
|
|
def clear_session(self, session_key: str):
|
|
"""Clear crisis state for a session."""
|
|
self._sessions.pop(session_key, None)
|
|
|
|
|
|
# Module-level tracker
|
|
_tracker = CrisisSessionTracker()
|
|
|
|
|
|
def check_crisis_in_gateway(session_key: str, message: str) -> Tuple[bool, Optional[str]]:
|
|
"""
|
|
Check message for crisis in gateway context.
|
|
|
|
This is the function called from gateway/run.py _handle_message.
|
|
Returns (should_block, crisis_response).
|
|
"""
|
|
is_crisis, response = _tracker.record(session_key, message)
|
|
return is_crisis, response
|
|
|
|
|
|
def notify_user_crisis_resources(session_key: str) -> str:
|
|
"""Get crisis resources for a session."""
|
|
return _CRISIS_RESPONSE
|
|
|
|
|
|
def is_crisis_session(session_key: str) -> bool:
|
|
"""Check if session is in crisis mode."""
|
|
return _tracker.is_crisis_session(session_key)
|