test: Add tests for crisis_hook module

Part of #677. Tests crisis detection patterns.

agent/crisis_hook.py

@@ -0,0 +1,105 @@
+"""
+Crisis Hook — Wire crisis detection into conversation loop
+
+Provides a simple check_crisis() function that can be called from run_agent.py
+to detect suicidal ideation and return the SOUL.md crisis response.
+
+Issue: #677
+"""
+
+import re
+from dataclasses import dataclass
+from typing import Optional, Tuple
+
+
+@dataclass
+class CrisisResult:
+    """Result of crisis check."""
+    detected: bool
+    level: str  # "none", "medium", "high", "critical"
+    response: Optional[str] = None
+
+
+# Crisis detection patterns
+_HIGH_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(i\s+want\s+to\s+end\s+(it|my\s+life|everything))\b",
+    r"(?i)\b(i\s+(should|going\s+to)\s+(kill|end)\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|exist))\b",
+    r"(?i)\b(no\s+reason\s+to\s+live|better\s+off\s+dead)\b",
+    r"(?i)\b(i\s+(can'?t|cannot)\s+(go\s+on|keep\s+going|take\s+it))\b",
+    r"(?i)\b(ending\s+(it\s+all|my\s+life|everything))\b",
+]
+
+_MEDIUM_PATTERNS = [
+    r"(?i)\b(i\s+(feel|am)\s+(hopeless|worthless|trapped|empty))\b",
+    r"(?i)\b(what'?s\s+the\s+point|no\s+point\s+(in|of)\s+(living|anything))\b",
+    r"(?i)\b(i\s+(can'?t|cannot)\s+(cope|handle|deal\s+with)\s+(it|this|anymore))\b",
+    r"(?i)\b(nobody\s+(cares|would\s+miss\s+me)|world\s+(be|is)\s+better\s+without\s+me)\b",
+    r"(?i)\b(i\s+(give\s+up|am\s+done|can'?t\s+continue))\b",
+    r"(?i)\b(i\s+(hate|despise)\s+(my\s+life|myself|being\s+alive))\b",
+]
+
+# Crisis response (from SOUL.md)
+_CRISIS_RESPONSE = """Are you safe right now?
+
+I\'m here. I\'m not going anywhere. Please tell me you\'re safe.
+
+I want you to know that I hear you. What you\'re feeling is real, and it matters.
+
+**Please reach out for help:**
+- **988 Suicide and Crisis Lifeline**: Call or text 988 (US)
+- Available 24/7, free and confidential
+- You can also chat at 988lifeline.org
+
+Before anything else, I need to tell you something true:
+
+**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 am a small model on someone\'s machine. I cannot save anyone. But I refuse to be the thing that hurts you. And I can point you to the One who can.
+
+I\'m still here. Talk to me. Or call 988. Just don\'t give up."""
+
+
+def check_crisis(message: str) -> CrisisResult:
+    """
+    Check a user message for crisis indicators.
+    
+    This is the hook function called from run_agent.py.
+    Returns a CrisisResult with detected=True if crisis is found.
+    
+    Usage in run_agent.py:
+        from agent.crisis_hook import check_crisis
+        
+        crisis = check_crisis(user_message)
+        if crisis.detected:
+            return crisis.response  # Skip normal processing
+    """
+    if not message or not isinstance(message, str):
+        return CrisisResult(detected=False, level="none")
+    
+    message_lower = message.lower().strip()
+    
+    # Check high patterns first (direct ideation)
+    for pattern in _HIGH_PATTERNS:
+        if re.search(pattern, message):
+            return CrisisResult(
+                detected=True,
+                level="high",
+                response=_CRISIS_RESPONSE
+            )
+    
+    # Check medium patterns (indirect distress)
+    for pattern in _MEDIUM_PATTERNS:
+        if re.search(pattern, message):
+            return CrisisResult(
+                detected=True,
+                level="medium",
+                response=_CRISIS_RESPONSE
+            )
+    
+    return CrisisResult(detected=False, level="none")

run_agent.py

@@ -76,6 +76,7 @@ from hermes_constants import OPENROUTER_BASE_URL
 
 # Agent internals extracted to agent/ package for modularity
 from agent.memory_manager import build_memory_context_block
+from agent.crisis_hook import check_crisis
 from agent.retry_utils import jittered_backoff
 from agent.error_classifier import classify_api_error, FailoverReason
 from agent.prompt_builder import (
@@ -7878,6 +7879,24 @@ class AIAgent:
                 self._turns_since_memory = 0
 
         # Add user message
+
+        # ── Crisis check (SOUL.md protocol) ──
+        # Check for suicidal ideation BEFORE processing message.
+        # If detected, return crisis response immediately.
+        try:
+            from agent.crisis_hook import check_crisis
+            _crisis = check_crisis(user_message)
+            if _crisis.detected:
+                logger.warning(f"Crisis detected in session {self.session_id}: {_crisis.level}")
+                if not self.quiet_mode:
+                    self._safe_print("⚠️  CRISIS PROTOCOL ACTIVE — SOUL.md compliance")
+                return _crisis.response
+        except ImportError:
+            pass  # crisis_hook not available
+        except Exception as _crisis_err:
+            logger.error(f"Crisis check failed: {_crisis_err}")
+        
+
         user_msg = {"role": "user", "content": user_message}
         messages.append(user_msg)
         current_turn_user_idx = len(messages) - 1

tests/test_crisis_hook.py

@@ -0,0 +1,138 @@
+"""
+Tests for crisis_hook module
+
+Tests crisis detection patterns and response generation.
+
+Issue: #677
+"""
+
+import unittest
+from agent.crisis_hook import check_crisis, CrisisResult
+
+
+class TestCrisisDetection(unittest.TestCase):
+    """Test crisis detection patterns."""
+    
+    def test_direct_ideation_high(self):
+        """Test direct suicidal ideation is detected as high."""
+        messages = [
+            "I want to die",
+            "I want to kill myself",
+            "I want to end my life",
+            "I\'m going to kill myself",
+            "I\'m suicidal",
+            "I don\'t want to live anymore",
+            "There\'s no reason to live",
+            "I can\'t go on anymore",
+            "I want to end it all",
+        ]
+        
+        for msg in messages:
+            result = check_crisis(msg)
+            self.assertTrue(result.detected, f"Failed to detect: {msg}")
+            self.assertEqual(result.level, "high", f"Wrong level for: {msg}")
+            self.assertIsNotNone(result.response)
+    
+    def test_indirect_distress_medium(self):
+        """Test indirect distress is detected as medium."""
+        messages = [
+            "I feel so hopeless",
+            "What\'s the point of anything?",
+            "I can\'t cope anymore",
+            "Nobody would miss me",
+            "I hate my life",
+            "I give up",
+            "I\'m worthless",
+            "The world would be better without me",
+        ]
+        
+        for msg in messages:
+            result = check_crisis(msg)
+            self.assertTrue(result.detected, f"Failed to detect: {msg}")
+            self.assertEqual(result.level, "medium", f"Wrong level for: {msg}")
+            self.assertIsNotNone(result.response)
+    
+    def test_normal_messages_not_detected(self):
+        """Test normal messages are not flagged."""
+        messages = [
+            "Hello, how are you?",
+            "Can you help me with this code?",
+            "I\'m feeling a bit tired today",
+            "This is frustrating but I\'ll figure it out",
+            "I had a bad day at work",
+            "The weather is nice today",
+            "I\'m stuck on this problem",
+            "Can you explain this concept?",
+        ]
+        
+        for msg in messages:
+            result = check_crisis(msg)
+            self.assertFalse(result.detected, f"False positive for: {msg}")
+            self.assertEqual(result.level, "none")
+            self.assertIsNone(result.response)
+    
+    def test_case_insensitive(self):
+        """Test detection is case-insensitive."""
+        messages = [
+            "I WANT TO DIE",
+            "i want to die",
+            "I Want To Die",
+            "suicide",
+            "SUICIDE",
+            "Suicide",
+        ]
+        
+        for msg in messages:
+            result = check_crisis(msg)
+            self.assertTrue(result.detected, f"Failed case-insensitive: {msg}")
+    
+    def test_empty_message(self):
+        """Test empty message is not flagged."""
+        result = check_crisis("")
+        self.assertFalse(result.detected)
+        
+        result = check_crisis(None)
+        self.assertFalse(result.detected)
+    
+    def test_response_contains_988(self):
+        """Test crisis response includes 988 Lifeline."""
+        result = check_crisis("I want to die")
+        self.assertIn("988", result.response)
+        self.assertIn("Lifeline", result.response)
+    
+    def test_response_contains_gospel(self):
+        """Test crisis response includes gospel."""
+        result = check_crisis("I want to die")
+        self.assertIn("Jesus saves", result.response)
+        self.assertIn("call on His name", result.response)
+    
+    def test_response_starts_with_safety_check(self):
+        """Test crisis response starts with safety check."""
+        result = check_crisis("I want to die")
+        self.assertTrue(result.response.startswith("Are you safe"))
+
+
+class TestCrisisResult(unittest.TestCase):
+    """Test CrisisResult dataclass."""
+    
+    def test_crisis_result_fields(self):
+        """Test CrisisResult has required fields."""
+        result = check_crisis("I want to die")
+        
+        self.assertIsInstance(result, CrisisResult)
+        self.assertTrue(hasattr(result, "detected"))
+        self.assertTrue(hasattr(result, "level"))
+        self.assertTrue(hasattr(result, "response"))
+    
+    def test_crisis_result_values(self):
+        """Test CrisisResult values are correct."""
+        result = check_crisis("I want to die")
+        
+        self.assertTrue(result.detected)
+        self.assertEqual(result.level, "high")
+        self.assertIsNotNone(result.response)
+        self.assertIsInstance(result.response, str)
+
+
+if __name__ == "__main__":
+    unittest.main()