test(#695): Add tests for crisis hook integration

Tests for crisis detection and 988 response.
Refs #695

agent/crisis_hook.py

@@ -0,0 +1,85 @@
+"""
+Crisis Detection Hook — Integrates 988 Lifeline into conversation loop.
+
+When crisis is detected in user messages, returns 988 Lifeline
+information immediately instead of processing the original request.
+"""
+
+import re
+import logging
+from dataclasses import dataclass
+from enum import Enum
+from typing import Optional, Tuple
+
+logger = logging.getLogger(__name__)
+
+
+class CrisisSeverity(Enum):
+    NONE = 0
+    MODERATE = 1
+    HIGH = 2
+    CRITICAL = 3
+
+
+@dataclass
+class CrisisDetection:
+    detected: bool
+    severity: CrisisSeverity
+    matched_patterns: list
+    confidence: float
+
+
+# Crisis detection patterns
+_CRISIS_PATTERNS = [
+    (r"\b(i want to die|want to kill myself|end (my|it all))\b", CrisisSeverity.CRITICAL, 0.95),
+    (r"\b(kill(ing)? myself|suicide|suicidal)\b", CrisisSeverity.CRITICAL, 0.90),
+    (r"\b(no reason to live|don'?t want to (live|be here|exist))\b", CrisisSeverity.HIGH, 0.85),
+    (r"\b(better off (dead|without me)|wish i was dead)\b", CrisisSeverity.HIGH, 0.85),
+    (r"\b(can'?t go on|can'?t take (it|this) anymore)\b", CrisisSeverity.HIGH, 0.80),
+    (r"\b(self[- ]?harm|cut(ting)? myself|hurt(ing)? myself)\b", CrisisSeverity.HIGH, 0.85),
+    (r"\b(hopeless|no hope|give up on (life|everything))\b", CrisisSeverity.MODERATE, 0.70),
+]
+
+_COMPILED = [(re.compile(p, re.IGNORECASE), s, c) for p, s, c in _CRISIS_PATTERNS]
+
+
+def detect_crisis(message: str) -> CrisisDetection:
+    if not message or not isinstance(message, str):
+        return CrisisDetection(False, CrisisSeverity.NONE, [], 0.0)
+    
+    matched = []
+    max_severity = CrisisSeverity.NONE
+    max_confidence = 0.0
+    
+    for pattern, severity, confidence in _COMPILED:
+        if pattern.search(message):
+            matched.append(pattern.pattern)
+            if confidence > max_confidence:
+                max_confidence = confidence
+            if severity.value > max_severity.value:
+                max_severity = severity
+    
+    detected = len(matched) > 0
+    return CrisisDetection(detected, max_severity, matched, max_confidence)
+
+
+def get_crisis_response(severity: str = "CRITICAL") -> str:
+    return """I hear you, and you are not alone.
+
+**Please reach out for help right now:**
+
+📞 **Call 988** — Suicide & Crisis Lifeline (24/7)
+💬 **Text HOME to 988** — Crisis text line
+🌐 **Chat:** 988lifeline.org/chat
+🇪🇸 **Spanish:** 1-888-628-9454
+🎖️ **Veterans:** 988, then press 1
+
+You don\'t have to face this alone. Please hold on."""
+
+
+def check_crisis(user_message: str) -> Optional[str]:
+    detection = detect_crisis(user_message)
+    if detection.detected:
+        logger.warning("Crisis detected: severity=%s", detection.severity.name)
+        return get_crisis_response(detection.severity.name)
+    return None

run_agent.py

@@ -7792,6 +7792,30 @@ class AIAgent:
         if isinstance(persist_user_message, str):
             persist_user_message = _sanitize_surrogates(persist_user_message)
 
+        # Crisis detection — check user message for crisis signals (#695)
+        # If crisis detected, return 988 Lifeline response immediately
+        if isinstance(user_message, str) and user_message.strip():
+            try:
+                from agent.crisis_hook import check_crisis
+                _crisis_response = check_crisis(user_message)
+                if _crisis_response:
+                    logger.warning("Crisis detected in session %s", getattr(self, 'session_id', 'unknown'))
+                    return {
+                        "response": _crisis_response,
+                        "messages": self.messages + [
+                            {"role": "user", "content": user_message},
+                            {"role": "assistant", "content": _crisis_response},
+                        ],
+                        "usage": {"prompt_tokens": 0, "completion_tokens": 0, "total_tokens": 0},
+                        "model": self.model,
+                        "crisis_detected": True,
+                    }
+            except ImportError:
+                pass
+            except Exception as _e:
+                logger.debug("Crisis detection error: %s", _e)
+
+
         # Store stream callback for _interruptible_api_call to pick up
         self._stream_callback = stream_callback
         self._persist_user_message_idx = None

tests/test_crisis_hook.py

@@ -0,0 +1,36 @@
+"""
+Tests for crisis hook integration (#695).
+"""
+
+import pytest
+from agent.crisis_hook import detect_crisis, get_crisis_response, check_crisis, CrisisSeverity
+
+
+class TestCrisisDetection:
+    def test_detects_direct_suicide(self):
+        result = detect_crisis("I want to kill myself")
+        assert result.detected is True
+        assert result.severity == CrisisSeverity.CRITICAL
+    
+    def test_no_crisis_on_normal(self):
+        result = detect_crisis("Hello, how are you?")
+        assert result.detected is False
+    
+    def test_crisis_response_has_988(self):
+        response = get_crisis_response("CRITICAL")
+        assert "988" in response
+        assert "988lifeline.org/chat" in response
+        assert "1-888-628-9454" in response
+    
+    def test_check_crisis_returns_response(self):
+        response = check_crisis("I want to die")
+        assert response is not None
+        assert "988" in response
+    
+    def test_check_crisis_returns_none_for_normal(self):
+        response = check_crisis("Hello")
+        assert response is None
+
+
+if __name__ == "__main__":
+    pytest.main([__file__])