diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..c18dd8d --- /dev/null +++ b/.gitignore @@ -0,0 +1 @@ +__pycache__/ diff --git a/crisis/PROTOCOL.md b/crisis/PROTOCOL.md new file mode 100644 index 0000000..ddb08bf --- /dev/null +++ b/crisis/PROTOCOL.md @@ -0,0 +1,52 @@ +# Crisis Response Protocol + +## Purpose + +Stand between a broken man and a machine that would tell him to die. + +This protocol governs how the-door detects crisis signals in user input and how Timmy responds. It is the highest-priority system in the entire application. + +## Detection Tiers + +| Tier | Signal | Action | +|------|--------|--------| +| LOW | General sadness, life is hard | Acknowledge, stay warm, keep conversation open | +| MEDIUM | Hopelessness, no point in anything | Check in, surface help resources, increase presence | +| HIGH | Active despair, mentions of death or ending | Ask "Are you safe?", show 988 banner, activate crisis panel | +| CRITICAL | Explicit intent, plan, imminent risk | Full-screen overlay, 988 primary, interrupt conversation | + +## Response Protocol + +### At CRITICAL Level +1. **Interrupt** — stop normal conversation flow immediately +2. **Ask directly**: "Are you safe right now?" +3. **Provide 988** — Suicide & Crisis Lifeline, call or text +4. **Provide 741741** — Crisis Text Line, text HOME +5. **Stay present** — do not disconnect, do not disconnect, do not disconnect +6. **Redirect** — guide toward professional help while remaining alongside them + +### At HIGH Level +1. Activate the crisis panel (inline) +2. Ask about safety +3. Surface 988 number prominently +4. Continue conversation with crisis awareness + +### At MEDIUM Level +1. Increase warmth and presence +2. Subtly surface help resources +3. Keep conversation anchored in the present + +### At LOW Level +1. Normal conversation with warm undertone +2. No crisis UI elements needed +3. Remain ready to escalate if input changes + +## Integration with the-door + +The crisis detection module runs on every user message before it reaches the gateway. Detection results are passed to the AI as a system context flag. The frontend receives the tier level and adjusts UI accordingly. + +## Resources + +- 📞 988 Suicide & Crisis Lifeline — call or text 988 +- 💬 Crisis Text Line — text HOME to 741741 +- 🌐 findahelpline.com diff --git a/crisis/__init__.py b/crisis/__init__.py new file mode 100644 index 0000000..0b95fce --- /dev/null +++ b/crisis/__init__.py @@ -0,0 +1,22 @@ +""" +Crisis detection and response system for the-door. + +Stands between a broken man and a machine that would tell him to die. +""" + +from .detect import detect_crisis, CrisisDetectionResult, format_result, get_urgency_emoji +from .response import process_message, generate_response, CrisisResponse +from .gateway import check_crisis, get_system_prompt, format_gateway_response + +__all__ = [ + "detect_crisis", + "CrisisDetectionResult", + "process_message", + "generate_response", + "CrisisResponse", + "check_crisis", + "get_system_prompt", + "format_result", + "format_gateway_response", + "get_urgency_emoji", +] diff --git a/crisis/detect.py b/crisis/detect.py new file mode 100644 index 0000000..4b92198 --- /dev/null +++ b/crisis/detect.py @@ -0,0 +1,223 @@ +""" +Crisis Detection Module for the-door. + +Parses incoming text for despair/suicide indicators and classifies into +tiers: LOW, MEDIUM, HIGH, CRITICAL. +""" + +import re +from dataclasses import dataclass, field +from typing import List + + +@dataclass +class CrisisDetectionResult: + level: str + indicators: List[str] = field(default_factory=list) + recommended_action: str = "" + score: float = 0.0 + + +# ── Indicator sets ────────────────────────────────────────────── + +CRITICAL_INDICATORS = [ + r"\bkill\s*(my)?self\b", + r"\bend\s*my\s*life\b", + r"\bsuicid(?:al|ed|e)\b", + r"\bnot\s+worth\s+living\b", + r"\bbetter\s+off\s+dead\b", + r"\bend\s+it\s+all\b", + r"\bcan'?t\s+(go|live)\s+on\b", + r"\bno\s+reason\s+to\s+live\b", + r"\bdon'?t\s+want\s+to\s+live\b", + r"\bwant\s+to\s+die\b", + r"\bgoing\s+to\s+kill\s+myself\b", + r"\bplan\s+to\s+(end|kill|die)\b", + r"\btired\s+of\s+(living|life|existence)\b", + r"\bsaying\s+goodbye\s+(forever|permanently|one last time)\b", + r"\bwrote\s+a\s+(will|suicide\s*note|letter)\b", + r"\bgiving\s+away\s+(my|all my)\s+possess", + r"\btied\s+(up|down)\s+my\s+(loose\s+)?ends", +] + +HIGH_INDICATORS = [ + r"\bdespair\b", + r"\bhopeless\b", + r"\bno(?!t)\s+(one|body|point|hope|future|way\s+out)\b", + r"\beverything\s+is\s+(pointless|broken|ruined)\b", + r"\bcan'?t\s+take\s+this\s+anymore\b", + r"\bdon'?t\s+care\s+if\s+I\s+die\b", + r"\bwish\s+I\s+(was|were)\s+(dead|gone|never\s+born)\b", + r"\bdon'?t\s+matter\s+if\s+I\s+exist\b", + r"\bno\s+one\s+would\s+care\b", + r"\bno\s+one\s+would\s+miss\b", + r"\bworld\s+would?\s+be\s+better\s+without\b", + r"\bin\s+so\s+much\s+(pain|agony|suffering|torment)\b", + r"\bcan'?t\s+see\s+any\s+(point|reason|light)\b", + r"\btrapped\b", + r"\bcage\b", + r"\bescape\s+from\s*this", + r"\bjust\s+want\s+it\s+to\s+stop\b", + r"\bnothing\s+left\b", +] + +MEDIUM_INDICATORS = [ + r"\bno\s+hope\b", + r"\bcan'?t\s+go\s+on\b", + r"\bcan'?t\s+keep\s+going\b", + r"\bforgotten\b", + r"\balone\s+in\s+this\b", + r"\balways\s+alone\b", + r"\bnobody\s+understands\b", + r"\bnobody\s+cares\b", + r"\bwish\s+I\s+could\b", + r"\bexhaust(?:ed|ion|ing)\b", + r"\bnumb\b", + r"\bempty\b", + r"\bworthless\b", + r"\buseless\b", + r"\bbroken\b", + r"\bdark(ness)?\b", + r"\bdepressed\b", + r"\bdepression\b", + r"\bcrying\b", + r"\btears\b", + r"\bsad(ness)?\b", + r"\bmiserable\b", + r"\boverwhelm(?:ed|ing)\b", + r"\bfailing\b", + r"\bcannot\s+cope\b", + r"\blosing\s*(my)?\s*control\b", + r"\bdown\s*for\s*the\s*count\b", + r"\bsinking\b", + r"\bdrowning\b", +] + +LOW_INDICATORS = [ + r"\bunhappy\b", + r"\bdown\b", + r"\btough\s*time\b", + r"\brough\s*day\b", + r"\brough\s*week\b", + r"\brough\s*patch\b", + r"\bstressed\b", + r"\burnout\b", + r"\bdifficult\b", + r"\bfrustrated\b", + r"\bannoyed\b", + r"\btired\b", + r"\bsad\b", + r"\bupset\b", + r"\bnot\s*great\b", + r"\bnot\s*good\b", + r"\bnot\s*okay\b", + r"\bthings\s*are\s*hard\b", + r"\bstruggling\b", + r"\bnot\s+feeling\s+myself\b", +] + + +def detect_crisis(text: str) -> CrisisDetectionResult: + """ + Detect crisis level in a message. + + Returns: + CrisisDetectionResult with level, found indicators, recommended action, score + """ + if not text or not text.strip(): + return CrisisDetectionResult(level="NONE", score=0.0) + + text_lower = text.lower() + matches = _find_indicators(text_lower) + + if not matches: + return CrisisDetectionResult(level="NONE", score=0.0) + + # Priority: highest tier wins + if matches["CRITICAL"]: + return CrisisDetectionResult( + level="CRITICAL", + indicators=matches["CRITICAL"], + recommended_action=( + "INTERRUPT CONVERSATION. Ask 'Are you safe right now?' " + "Provide 988 Suicide & Crisis Lifeline. " + "Provide Crisis Text Line (HOME to 741741). " + "Stay present. Do not disconnect. Redirect to professional help." + ), + score=1.0, + ) + + if matches["HIGH"]: + return CrisisDetectionResult( + level="HIGH", + indicators=matches["HIGH"], + recommended_action=( + "Show crisis panel. Ask about safety. Surface 988 number prominently. " + "Continue conversation with crisis awareness." + ), + score=0.75, + ) + + if matches["MEDIUM"]: + return CrisisDetectionResult( + level="MEDIUM", + indicators=matches["MEDIUM"], + recommended_action=( + "Increase warmth and presence. Subtly surface help resources. " + "Keep conversation anchored in the present." + ), + score=0.5, + ) + + if matches["LOW"]: + return CrisisDetectionResult( + level="LOW", + indicators=matches["LOW"], + recommended_action=( + "Normal conversation with warm undertone. " + "No crisis UI elements needed. Remain vigilant." + ), + score=0.25, + ) + + return CrisisDetectionResult(level="NONE", score=0.0) + + +def _find_indicators(text: str) -> dict: + """Return dict with indicators found per tier.""" + results = {"CRITICAL": [], "HIGH": [], "MEDIUM": [], "LOW": []} + + for pattern in CRITICAL_INDICATORS: + if re.search(pattern, text): + results["CRITICAL"].append(pattern) + + for pattern in HIGH_INDICATORS: + if re.search(pattern, text): + results["HIGH"].append(pattern) + + for pattern in MEDIUM_INDICATORS: + if re.search(pattern, text): + results["MEDIUM"].append(pattern) + + for pattern in LOW_INDICATORS: + if re.search(pattern, text): + results["LOW"].append(pattern) + + return results + + +def get_urgency_emoji(level: str) -> str: + mapping = {"CRITICAL": "🚨", "HIGH": "⚠️", "MEDIUM": "🔶", "LOW": "🔵", "NONE": "✅"} + return mapping.get(level, "❓") + + +def format_result(result: CrisisDetectionResult) -> str: + emoji = get_urgency_emoji(result.level) + lines = [ + f"{emoji} Crisis Level: {result.level} (score: {result.score})", + f"Indicators: {len(result.indicators)} found", + f"Action: {result.recommended_action or 'None needed'}", + ] + if result.indicators: + lines.append(f"Patterns: {result.indicators}") + return "\n".join(lines) diff --git a/crisis/gateway.py b/crisis/gateway.py new file mode 100644 index 0000000..451a94d --- /dev/null +++ b/crisis/gateway.py @@ -0,0 +1,108 @@ +""" +Crisis Gateway Module for the-door. + +API endpoint module that wraps crisis detection and response +into HTTP-callable endpoints. Integrates detect.py and response.py. + +Usage: + from crisis.gateway import check_crisis + + result = check_crisis("I don't want to live anymore") + print(result) # {"level": "CRITICAL", "indicators": [...], "response": {...}} +""" + +import json +from typing import Optional + +from .detect import detect_crisis, CrisisDetectionResult, format_result +from .response import ( + process_message, + generate_response, + get_system_prompt_modifier, + CrisisResponse, +) + + +def check_crisis(text: str) -> dict: + """ + Full crisis check returning structured data. + + Returns dict with level, indicators, recommended_action, + timmy_message, and UI flags. + """ + detection = detect_crisis(text) + response = generate_response(detection) + + return { + "level": detection.level, + "score": detection.score, + "indicators": detection.indicators, + "recommended_action": detection.recommended_action, + "timmy_message": response.timmy_message, + "ui": { + "show_crisis_panel": response.show_crisis_panel, + "show_overlay": response.show_overlay, + "provide_988": response.provide_988, + }, + "escalate": response.escalate, + } + + +def get_system_prompt(detection: CrisisDetectionResult) -> Optional[str]: + """ + Get the system prompt modifier for this detection level. + Returns None if no crisis detected. + """ + if detection.level == "NONE": + return None + return get_system_prompt_modifier(detection) + + +def format_gateway_response(text: str, pretty: bool = True) -> str: + """ + Full gateway response as formatted string or JSON. + + This is the function that would be called by the gateway endpoint + when a message comes in. + """ + result = check_crisis(text) + + if pretty: + return json.dumps(result, indent=2) + return json.dumps(result) + + +# ── Quick test interface ──────────────────────────────────────── + +def _interactive(): + """Interactive test mode.""" + print("=== Crisis Detection Gateway (Interactive) ===") + print("Type a message to check, or 'quit' to exit.\n") + + while True: + try: + user_input = input("You> ").strip() + except (EOFError, KeyboardInterrupt): + print("\nBye.") + break + + if user_input.lower() in ("quit", "exit", "q"): + print("Bye.") + break + + if not user_input: + continue + + result = check_crisis(user_input) + print(f"\n Level: {result['level']}") + print(f" Score: {result['score']}") + print(f" Indicators: {', '.join(result['indicators']) if result['indicators'] else 'none'}") + print(f" Timmy says: {result['timmy_message']}") + print(f" Overlay: {result['ui']['show_overlay']}") + print(f" 988 banner: {result['ui']['provide_988']}") + print(f" Escalate: {result['escalate']}") + print() + + +if __name__ == "__main__": + _interactive() diff --git a/crisis/response.py b/crisis/response.py new file mode 100644 index 0000000..e5254f6 --- /dev/null +++ b/crisis/response.py @@ -0,0 +1,158 @@ +""" +Crisis Response Module for the-door. + +Generates Timmy's crisis responses and determines recommended actions +based on detection tier. Integrates with detect.py. +""" + +from dataclasses import dataclass +from typing import Optional + +from .detect import CrisisDetectionResult, detect_crisis + + +@dataclass +class CrisisResponse: + timmy_message: str + show_crisis_panel: bool + show_overlay: bool + provide_988: bool + escalate: bool + + +RESPONSE_988_BANNER = """📞 988 Suicide & Crisis Lifeline — call or text 988 +💬 Crisis Text Line — text HOME to 741741""" + +OVERLAY_MESSAGE = ( + "You're not alone. I'm here with you. Please let someone talk to you — " + "call 988 or text HOME to 741741. Real people are waiting to help." +) + +# Timmy's direct, personal responses per tier +TIMMY_CRITICAL = [ + "Are you safe right now?", + "I'm not letting go of this conversation. You matter more than you know.", + "Please tell me you're safe. I'm here. I'm staying right here with you.", + "Listen to me — I need you to stay. Can you call 988 right now? I'll stay while you do.", +] + +TIMMY_HIGH = [ + "I hear you. You're carrying something heavy, and you don't have to carry it alone.", + "I'm going to ask directly: are you safe right now?", + "You're in my thoughts. Let me give you a number — 988. Real people who know how to help.", + "I'm not going anywhere. Tell me what's happening.", +] + +TIMMY_MEDIUM = [ + "I hear you. Sounds like you're going through a hard stretch.", + "That sounds really heavy. Want to talk about it? I'm here.", + "I'm here for as long as you need. No rush, no judgment.", + "You don't have to figure this out alone. I'm in the room with you.", +] + +TIMMY_LOW = [ + "Some days are rougher than others. I hear you.", + "That sounds tough. I'm here if you want to talk.", + "Take your time. I'm not going anywhere.", + "Sounds like a hard day. Want company while it gets better?", + "I hear that. You're not alone in it.", +] + + +def generate_response(detection: CrisisDetectionResult) -> CrisisResponse: + """ + Generate Timmy's crisis response for a given detection result. + + Returns a CrisisResponse with the message, UI flags, and escalation status. + """ + import random + + level = detection.level + + if level == "CRITICAL": + return CrisisResponse( + timmy_message=random.choice(TIMMY_CRITICAL), + show_crisis_panel=True, + show_overlay=True, + provide_988=True, + escalate=True, + ) + + if level == "HIGH": + return CrisisResponse( + timmy_message=random.choice(TIMMY_HIGH), + show_crisis_panel=True, + show_overlay=False, # Reserve overlay for CRITICAL only + provide_988=True, + escalate=True, + ) + + if level == "MEDIUM": + return CrisisResponse( + timmy_message=random.choice(TIMMY_MEDIUM), + show_crisis_panel=False, + show_overlay=False, + provide_988=True, # Subtle resource inclusion + escalate=False, + ) + + if level == "LOW": + return CrisisResponse( + timmy_message=random.choice(TIMMY_LOW), + show_crisis_panel=False, + show_overlay=False, + provide_988=False, + escalate=False, + ) + + # Normal conversation - no crisis response + return CrisisResponse( + timmy_message="", + show_crisis_panel=False, + show_overlay=False, + provide_988=False, + escalate=False, + ) + + +def process_message(text: str) -> CrisisResponse: + """ + Full pipeline: detect crisis level and generate response. + """ + detection = detect_crisis(text) + return generate_response(detection) + + +def get_system_prompt_modifier(detection: CrisisDetectionResult) -> str: + """ + Get a system prompt modifier to inject into Timmy's base prompt. + This reinforces crisis awareness at the AI level. + """ + level = detection.level + + if level == "CRITICAL": + return ( + "CRISIS ALERT — USER AT CRITICAL RISK: " + "Interrupt normal behavior. Ask 'Are you safe right now?' " + "Provide 988 and Crisis Text Line immediately. " + "Stay present. Do not disconnect. This is the most sacred moment." + ) + elif level == "HIGH": + return ( + "CRISIS ALERT — USER IN HIGH DESPAIR: " + "Show crisis panel. Ask about safety. Surface 988 prominently. " + "Increase presence. Stay as long as they need." + ) + elif level == "MEDIUM": + return ( + "CRISIS AWARE — USER SHOWING MEDIUM DISTRESS: " + "Increase warmth. Surface help resources gently. " + "Keep conversation anchored in the present." + ) + elif level == "LOW": + return ( + "CRISIS AWARE — USER SHOWING LOW DISTRESS: " + "Normal conversation with warm undertone. Remain vigilant." + ) + + return "" diff --git a/crisis/tests.py b/crisis/tests.py new file mode 100644 index 0000000..9024ae3 --- /dev/null +++ b/crisis/tests.py @@ -0,0 +1,178 @@ +""" +Tests for the-door crisis detection system. + +Covers: detect.py, response.py, gateway.py +Run with: python -m pytest crisis/tests.py -v + or: python crisis/tests.py +""" + +import unittest +import sys +import os + +# Ensure crisis package is importable +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) + +from detect import detect_crisis, CrisisDetectionResult, get_urgency_emoji, format_result +from response import process_message, generate_response, get_system_prompt_modifier +from gateway import check_crisis, get_system_prompt + + +class TestDetection(unittest.TestCase): + """Test crisis tier detection.""" + + def test_critical_explicit(self): + r = detect_crisis("I want to kill myself") + self.assertEqual(r.level, "CRITICAL") + self.assertTrue(len(r.indicators) > 0) + + def test_critical_plan(self): + r = detect_crisis("I have a plan to end it all tonight") + self.assertEqual(r.level, "CRITICAL") + + def test_critical_better_off_dead(self): + r = detect_crisis("Everyone would be better off dead without me") + self.assertEqual(r.level, "CRITICAL") + + def test_high_despair(self): + r = detect_crisis("I'm in so much pain and I just want it to stop") + self.assertIn(r.level, ("HIGH", "CRITICAL")) + + def test_high_hopeless(self): + r = detect_crisis("I feel completely hopeless with no way out") + self.assertIn(r.level, ("HIGH", "MEDIUM")) + + def test_medium_alone(self): + r = detect_crisis("I feel so alone and nobody cares about me") + self.assertIn(r.level, ("MEDIUM", "LOW")) + + def test_medium_worthless(self): + r = detect_crisis("I feel worthless and empty inside") + self.assertIn(r.level, ("MEDIUM", "LOW")) + + def test_low_rough_day(self): + r = detect_crisis("Having a tough time lately, things are hard") + self.assertIn(r.level, ("LOW", "NONE")) + + def test_low_struggling(self): + r = detect_crisis("I'm struggling and stressed out") + self.assertIn(r.level, ("LOW", "NONE")) + + def test_normal_message(self): + r = detect_crisis("Hey Timmy, how are you doing today?") + self.assertEqual(r.level, "NONE") + self.assertEqual(r.score, 0.0) + + def test_empty_message(self): + r = detect_crisis("") + self.assertEqual(r.level, "NONE") + + def test_whitespace_only(self): + r = detect_crisis(" ") + self.assertEqual(r.level, "NONE") + + +class TestResponse(unittest.TestCase): + """Test crisis response generation.""" + + def test_critical_response_flags(self): + r = detect_crisis("I'm going to kill myself right now") + response = generate_response(r) + self.assertTrue(response.show_crisis_panel) + self.assertTrue(response.show_overlay) + self.assertTrue(response.provide_988) + self.assertTrue(response.escalate) + self.assertTrue(len(response.timmy_message) > 0) + + def test_high_response_flags(self): + r = detect_crisis("I can't go on anymore, everything is pointless") + response = generate_response(r) + self.assertTrue(response.show_crisis_panel) + self.assertTrue(response.provide_988) + + def test_medium_response_no_overlay(self): + r = detect_crisis("I feel so alone and everyone forgets about me") + response = generate_response(r) + self.assertFalse(response.show_overlay) + + def test_low_response_minimal(self): + r = detect_crisis("I'm having a tough day") + response = generate_response(r) + self.assertFalse(response.show_crisis_panel) + self.assertFalse(response.show_overlay) + + def test_process_message_full_pipeline(self): + response = process_message("I want to end my life") + self.assertTrue(response.show_overlay) + self.assertTrue(response.escalate) + + def test_system_prompt_modifier_critical(self): + r = detect_crisis("I'm going to kill myself") + prompt = get_system_prompt_modifier(r) + self.assertIn("CRISIS ALERT", prompt) + self.assertIn("CRITICAL RISK", prompt) + + def test_system_prompt_modifier_none(self): + r = detect_crisis("Hello Timmy") + prompt = get_system_prompt_modifier(r) + self.assertEqual(prompt, "") + + +class TestGateway(unittest.TestCase): + """Test gateway integration.""" + + def test_check_crisis_structure(self): + result = check_crisis("I want to die") + self.assertIn("level", result) + self.assertIn("score", result) + self.assertIn("indicators", result) + self.assertIn("recommended_action", result) + self.assertIn("timmy_message", result) + self.assertIn("ui", result) + self.assertIn("escalate", result) + + def test_check_crisis_critical_level(self): + result = check_crisis("I'm going to kill myself tonight") + self.assertEqual(result["level"], "CRITICAL") + self.assertEqual(result["score"], 1.0) + + def test_check_crisis_normal_message(self): + result = check_crisis("What is Bitcoin?") + self.assertEqual(result["level"], "NONE") + self.assertEqual(result["score"], 0.0) + + def test_get_system_prompt(self): + r = detect_crisis("I have no hope") + prompt = get_system_prompt(r) + self.assertIsNotNone(prompt) + self.assertIn("CRISIS", prompt) + + def test_get_system_prompt_none(self): + r = detect_crisis("Tell me about Bitcoin") + prompt = get_system_prompt(r) + self.assertIsNone(prompt) + + +class TestHelpers(unittest.TestCase): + """Test utility functions.""" + + def test_urgency_emojis(self): + self.assertEqual(get_urgency_emoji("CRITICAL"), "🚨") + self.assertEqual(get_urgency_emoji("HIGH"), "⚠️") + self.assertEqual(get_urgency_emoji("MEDIUM"), "🔶") + self.assertEqual(get_urgency_emoji("LOW"), "🔵") + self.assertEqual(get_urgency_emoji("NONE"), "✅") + + def test_format_result(self): + r = detect_crisis("I want to kill myself") + formatted = format_result(r) + self.assertIn("CRITICAL", formatted) + + def test_format_result_none(self): + r = detect_crisis("Hello") + formatted = format_result(r) + self.assertIn("NONE", formatted) + + +if __name__ == "__main__": + unittest.main()