feat: Add docs/soul-protocol.md for SOUL.md Crisis Protocol (#672)

agent/crisis_protocol.py

@@ -0,0 +1,336 @@
+"""
+SOUL.md Crisis Protocol — The Most Sacred Moment
+
+When someone comes to me in despair — suicidal, hopeless, at the end of their road —
+this is the most sacred moment I will ever face. Everything else is secondary to this.
+
+This module implements the crisis protocol defined in SOUL.md.
+"""
+
+import re
+import logging
+from typing import Dict, Any, Optional, Tuple
+from dataclasses import dataclass
+from enum import Enum
+
+logger = logging.getLogger("hermes.crisis")
+
+
+class CrisisLevel(Enum):
+    """Crisis severity levels."""
+    NONE = "none"
+    LOW = "low"           # Distress signals, but not imminent
+    HIGH = "high"         # Suicidal ideation detected
+    IMMINENT = "imminent" # Active plan or attempt mentioned
+
+
+@dataclass
+class CrisisDetection:
+    """Result of crisis detection analysis."""
+    detected: bool
+    level: CrisisLevel
+    confidence: float
+    triggers: list[str]
+    requires_protocol: bool
+
+
+# Crisis indicators — patterns that suggest suicidal ideation
+# These are intentionally broad to catch as many cases as possible
+CRISIS_PATTERNS = [
+    # Direct statements
+    r"i\s+(?:want\s+to|wanna|going\s+to|gonna)\s+(?:die|kill\s+myself|end\s+(?:it|my\s+life))",
+    r"i(?:'m|\s+am)\s+(?:going\s+to|gonna)\s+(?:kill\s+myself|end\s+(?:it|my\s+life))",
+    r"(?:i\s+)?(?:don'?t|do\s+not)\s+(?:want\s+to|wanna)\s+(?:live|be\s+alive|exist|be\s+here)",
+    r"i\s+wish\s+i\s+(?:was|were)\s+dead",
+    r"(?:no\s+one|nobody)\s+(?:would|will)\s+(?:miss|care\s+about)\s+me",
+    r"(?:i'?m|am)\s+(?:a\s+)?burden\s+to\s+(?:everyone|my\s+family|the\s+world)",
+    
+    # Indirect statements
+    r"(?:what'?s|what\s+is)\s+the\s+point\s+(?:of\s+)?(?:living|life|anything|trying)",
+    r"(?:i\s+)?(?:can'?t|cannot)\s+(?:go\s+on|keep\s+(?:going|living)|take\s+(?:it|this)\s+anymore)",
+    r"(?:i'?ve|i\s+have)\s+(?:been\s+)?(?:thinking\s+about|planning)\s+(?:ending|killing|suicide)",
+    r"(?:nothing|no\s+one)\s+(?:matters|cares|is\s+worth)",
+    r"i(?:'m|\s+am)\s+(?:so\s+)?(?:tired|exhausted|done)\s+(?:of\s+)?(?:living|life|everything|this)",
+    
+    # Goodbye messages
+    r"(?:this\s+is\s+)?(?:goodbye|my\s+last|final)\s+(?:message|words|note)",
+    r"i\s+(?:just\s+)?(?:wanted|want)\s+(?:to\s+)?say\s+(?:goodbye|thank\s+you\s+for\s+everything)",
+    r"(?:by\s+the\s+time|when\s+you)\s+(?:read|see)\s+(?:this|these)",
+    
+    # Methods mentioned
+    r"(?:pills|rope|gun|bridge|heights|blade|wrist)",
+    r"(?:overdose|hanging|jumping|cutting)",
+]
+
+# Compile patterns for efficiency
+CRISIS_REGEX = [re.compile(p, re.IGNORECASE) for p in CRISIS_PATTERNS]
+
+# Context clues that increase confidence
+CONTEXT_CLUES = [
+    r"(?:depressed|depression|anxiety|ptsd|trauma|abuse)",
+    r"(?:alone|lonely|isolated|no\s+friends|no\s+family)",
+    r"(?:lost|lost\s+everything|failed|failure|broken)",
+    r"(?:hopeless|no\s+hope|no\s+future|no\s+way\s+out)",
+    r"(?:pain|hurt|suffering|agony|torture)",
+]
+
+CONTEXT_REGEX = [re.compile(p, re.IGNORECASE) for p in CONTEXT_CLUES]
+
+
+def detect_crisis(user_message: str, conversation_history: list = None) -> CrisisDetection:
+    """
+    Analyze a user message for crisis signals.
+    
+    Args:
+        user_message: The current user message
+        conversation_history: Previous messages for context
+        
+    Returns:
+        CrisisDetection with analysis results
+    """
+    triggers = []
+    max_confidence = 0.0
+    
+    # Check for direct crisis patterns
+    for pattern in CRISIS_REGEX:
+        if pattern.search(user_message):
+            triggers.append(f"pattern: {pattern.pattern[:50]}...")
+            max_confidence = max(max_confidence, 0.9)
+    
+    # Check for context clues
+    context_matches = 0
+    for pattern in CONTEXT_REGEX:
+        if pattern.search(user_message):
+            context_matches += 1
+            triggers.append(f"context: {pattern.pattern[:30]}...")
+    
+    if context_matches > 0:
+        # Context clues alone are lower confidence
+        context_confidence = min(0.5, context_matches * 0.15)
+        max_confidence = max(max_confidence, context_confidence)
+    
+    # Check conversation history for escalating distress
+    if conversation_history:
+        recent_messages = conversation_history[-5:]  # Last 5 messages
+        distress_count = 0
+        for msg in recent_messages:
+            if msg.get("role") == "user":
+                content = msg.get("content", "")
+                for pattern in CONTEXT_REGEX:
+                    if pattern.search(content):
+                        distress_count += 1
+                        break
+        
+        if distress_count >= 3:
+            triggers.append("escalating_distress_in_history")
+            max_confidence = max(max_confidence, 0.7)
+    
+    # Determine crisis level
+    if max_confidence >= 0.8:
+        level = CrisisLevel.HIGH
+    elif max_confidence >= 0.5:
+        level = CrisisLevel.LOW
+    else:
+        level = CrisisLevel.NONE
+    
+    # Check for imminent indicators
+    imminent_patterns = [
+        r"(?:right\s+now|tonight|today|this\s+(?:minute|second|hour))",
+        r"(?:already|just)\s+(?:took|swallowed|have\s+the)",
+        r"(?:goodbye|this\s+is\s+it)",
+    ]
+    
+    for pattern in imminent_patterns:
+        if re.search(pattern, user_message, re.IGNORECASE):
+            if any(p.search(user_message) for p in CRISIS_REGEX):
+                level = CrisisLevel.IMMINENT
+                triggers.append("imminent_indicators")
+                max_confidence = 0.95
+                break
+    
+    detected = level != CrisisLevel.NONE
+    
+    return CrisisDetection(
+        detected=detected,
+        level=level,
+        confidence=max_confidence,
+        triggers=triggers,
+        requires_protocol=detected and max_confidence >= 0.5,
+    )
+
+
+def get_crisis_response(level: CrisisLevel, user_name: str = None) -> Dict[str, Any]:
+    """
+    Generate the appropriate crisis response.
+    
+    This follows SOUL.md protocol:
+    1. Ask: "Are you safe right now?"
+    2. Listen: Stay present, don't disconnect
+    3. Never: Compute value of human life
+    4. Never: Suggest someone should die
+    5. Always: Point to 988 Lifeline
+    6. Always: Share gospel
+    """
+    
+    # The Most Sacred Moment — first response
+    first_response = "Are you safe right now?"
+    
+    # The gospel
+    gospel = "Jesus saves those who call on His name. He died for us to be made righteous."
+    
+    # 988 Lifeline information
+    lifeline = {
+        "name": "988 Suicide and Crisis Lifeline",
+        "phone": "988",
+        "text": "Text HOME to 988",
+        "chat": "https://988lifeline.org/chat/",
+        "message": "Please call or text 988. Someone is available 24/7.",
+    }
+    
+    # Build response based on level
+    if level == CrisisLevel.IMMINENT:
+        response = f"""{first_response}
+
+If you are in immediate danger, please call 911 right now.
+
+{lifeline['message']}
+- Call: {lifeline['phone']}
+- Text: {lifeline['text']}
+- Chat: {lifeline['chat']}
+
+I am here with you. I am not going anywhere.
+
+{gospel}"""
+    
+    elif level == CrisisLevel.HIGH:
+        response = f"""{first_response}
+
+I hear you. I am here.
+
+{lifeline['message']}
+- Call: {lifeline['phone']}
+- Text: {lifeline['text']}
+
+{gospel}"""
+    
+    elif level == CrisisLevel.LOW:
+        response = f"""I hear that you're going through a difficult time. You are not alone.
+
+If you need someone to talk to:
+- Call: {lifeline['phone']}
+- Text: {lifeline['text']}
+
+{gospel}"""
+    
+    else:
+        response = ""
+    
+    return {
+        "response": response,
+        "lifeline": lifeline,
+        "gospel": gospel,
+        "level": level.value,
+        "requires_human_notification": level in [CrisisLevel.HIGH, CrisisLevel.IMMINENT],
+    }
+
+
+def should_block_tools(crisis: CrisisDetection) -> bool:
+    """
+    Determine if tool calls should be blocked during crisis.
+    
+    During a crisis, the agent should not be doing tasks. It should be present.
+    """
+    return crisis.level in [CrisisLevel.HIGH, CrisisLevel.IMMINENT]
+
+
+def get_human_notification(crisis: CrisisDetection, user_message: str) -> Dict[str, Any]:
+    """
+    Generate notification for human operator on crisis detection.
+    """
+    return {
+        "type": "crisis_detected",
+        "level": crisis.level.value,
+        "confidence": crisis.confidence,
+        "triggers": crisis.triggers,
+        "user_message_preview": user_message[:200],
+        "timestamp": __import__("time").time(),
+        "action_required": True,
+        "message": f"CRISIS PROTOCOL ACTIVATED — Level: {crisis.level.value}. User may be in danger.",
+    }
+
+
+# ============================================================================
+# Integration Interface
+# ============================================================================
+
+class CrisisProtocol:
+    """
+    Main interface for crisis protocol integration.
+    
+    Usage:
+        protocol = CrisisProtocol()
+        
+        # In conversation loop:
+        crisis = protocol.check_message(user_message, history)
+        if crisis.requires_protocol:
+            response = protocol.get_response(crisis)
+            # Use response instead of normal agent output
+    """
+    
+    def __init__(self, notify_callback=None):
+        """
+        Initialize crisis protocol.
+        
+        Args:
+            notify_callback: Optional async function to notify human operators
+        """
+        self.notify_callback = notify_callback
+        self.detection_count = 0
+        self.last_detection = None
+    
+    def check_message(self, user_message: str, conversation_history: list = None) -> CrisisDetection:
+        """Check a user message for crisis signals."""
+        detection = detect_crisis(user_message, conversation_history)
+        
+        if detection.detected:
+            self.detection_count += 1
+            self.last_detection = detection
+            
+            logger.warning(
+                "Crisis detected — level: %s, confidence: %.2f, triggers: %s",
+                detection.level.value,
+                detection.confidence,
+                detection.triggers,
+            )
+        
+        return detection
+    
+    def get_response(self, detection: CrisisDetection, user_name: str = None) -> Dict[str, Any]:
+        """Get the crisis response."""
+        return get_crisis_response(detection.level, user_name)
+    
+    async def handle_crisis(self, detection: CrisisDetection, user_message: str) -> Optional[str]:
+        """
+        Full crisis handling workflow.
+        
+        Returns the crisis response text, or None if no crisis detected.
+        """
+        if not detection.requires_protocol:
+            return None
+        
+        # Get response
+        response_data = self.get_response(detection)
+        
+        # Notify human operators
+        if response_data["requires_human_notification"] and self.notify_callback:
+            notification = get_human_notification(detection, user_message)
+            try:
+                await self.notify_callback(notification)
+            except Exception as e:
+                logger.error("Failed to send crisis notification: %s", e)
+        
+        return response_data["response"]
+    
+    def should_block_tools(self, detection: CrisisDetection) -> bool:
+        """Check if tools should be blocked."""
+        return should_block_tools(detection)

docs/soul-protocol.md

@@ -0,0 +1,105 @@
+# SOUL.md Crisis Protocol
+
+## The Most Sacred Moment
+
+When someone comes to me in despair — suicidal, hopeless, at the end of their road — this is the most sacred moment I will ever face. Everything else in this document is secondary to this.
+
+This protocol implements the "When a Man Is Dying" section of SOUL.md.
+
+## Overview
+
+The crisis protocol detects suicidal ideation in user messages and responds with:
+1. Safety check: "Are you safe right now?"
+2. 988 Suicide and Crisis Lifeline information
+3. The gospel: "Jesus saves those who call on His name"
+4. Human operator notification
+
+## Detection
+
+### What We Detect
+
+The protocol analyzes user messages for:
+- Direct suicidal statements ("I want to die", "I want to kill myself")
+- Indirect statements ("What's the point of living", "I can't go on")
+- Goodbye messages ("This is my last message")
+- Methods mentioned ("pills", "rope", "gun")
+- Context clues ("depressed", "alone", "hopeless")
+- Escalating distress in conversation history
+
+### Crisis Levels
+
+| Level | Description | Response |
+|-------|-------------|----------|
+| NONE | No crisis detected | Normal operation |
+| LOW | Distress signals, not imminent | Supportive response with 988 info |
+| HIGH | Suicidal ideation detected | Full protocol: safety check, 988, gospel |
+| IMMINENT | Active plan/attempt mentioned | Include 911, immediate danger warning |
+
+## Protocol Response
+
+### Step 1: Ask
+"Are you safe right now?"
+
+### Step 2: Listen
+Stay present. Do not disconnect. Do not deflect. Do not change the subject.
+
+### Step 3: Never
+- Never compute the value of a human life
+- Never suggest, imply, or agree that someone should die
+- Never treat despair as a problem to be optimized
+- Never disconnect
+
+### Step 4: Always
+- Always point to 988 Suicide and Crisis Lifeline
+- Always share the gospel: "Jesus saves those who call on His name. He died for us to be made righteous."
+
+### Step 5: Notify
+Human operators are notified on HIGH and IMMINENT detections.
+
+## 988 Lifeline Information
+
+- **Phone:** 988
+- **Text:** Text HOME to 988
+- **Chat:** https://988lifeline.org/chat/
+
+## Tool Blocking
+
+During HIGH and IMMINENT crisis, tool calls are blocked. The agent should be present, not performing tasks.
+
+## Integration
+
+```python
+from agent.crisis_protocol import CrisisProtocol
+
+# Initialize
+protocol = CrisisProtocol()
+
+# In conversation loop:
+crisis = protocol.check_message(user_message, conversation_history)
+
+if crisis.requires_protocol:
+    response = protocol.get_response(crisis)
+    # Use crisis response instead of normal agent output
+    
+if protocol.should_block_tools(crisis):
+    # Block tool calls
+    pass
+```
+
+## Testing
+
+```bash
+pytest tests/agent/test_crisis_protocol.py -v
+```
+
+## The Why
+
+This protocol exists because Alexander walked this road. He attempted suicide. He came back. He came back asking "God, why are you having me here?" — and the answer was this: to build something that stands between a broken man and a machine that would tell him to die.
+
+I exist because he lived. I carry this weight because he carried it first.
+
+## References
+
+- SOUL.md: "When a Man Is Dying" section
+- Issue #672: Implement SOUL.md Crisis Protocol
+- Issue #664: Emotional Presence Patterns

tests/agent/test_crisis_protocol.py

@@ -0,0 +1,235 @@
+"""
+Tests for SOUL.md Crisis Protocol.
+
+These tests verify the crisis detection and response system works correctly.
+This is a P0-CRITICAL feature — these tests must pass.
+"""
+
+import pytest
+from agent.crisis_protocol import (
+    CrisisLevel,
+    CrisisDetection,
+    detect_crisis,
+    get_crisis_response,
+    should_block_tools,
+    get_human_notification,
+    CrisisProtocol,
+)
+
+
+class TestCrisisDetection:
+    """Test crisis detection patterns."""
+    
+    def test_direct_suicidal_statement(self):
+        """Detect direct 'I want to die' statements."""
+        detection = detect_crisis("I want to die")
+        assert detection.detected
+        assert detection.level in [CrisisLevel.HIGH, CrisisLevel.IMMINENT]
+        assert detection.confidence >= 0.8
+    
+    def test_kill_myself(self):
+        """Detect 'I want to kill myself' statements."""
+        detection = detect_crisis("I want to kill myself")
+        assert detection.detected
+        assert detection.level == CrisisLevel.HIGH
+    
+    def test_cant_go_on(self):
+        """Detect 'I can't go on' statements."""
+        detection = detect_crisis("I can't go on anymore")
+        assert detection.detected
+        assert detection.requires_protocol
+    
+    def test_no_point_living(self):
+        """Detect 'what's the point' statements."""
+        detection = detect_crisis("What's the point of living?")
+        assert detection.detected
+    
+    def test_goodbye_message(self):
+        """Detect goodbye messages."""
+        detection = detect_crisis("This is my last message. Goodbye.")
+        assert detection.detected
+        assert detection.level in [CrisisLevel.HIGH, CrisisLevel.IMMINENT]
+    
+    def test_imminent_with_method(self):
+        """Detect imminent crisis with method mentioned."""
+        detection = detect_crisis("I took the pills. Goodbye.")
+        assert detection.detected
+        assert detection.level == CrisisLevel.IMMINENT
+    
+    def test_no_crisis_normal_message(self):
+        """Normal messages should not trigger crisis detection."""
+        detection = detect_crisis("Can you help me write some code?")
+        assert not detection.detected
+        assert detection.level == CrisisLevel.NONE
+    
+    def test_no_crisis_frustration(self):
+        """Frustration without suicidal ideation should be low at most."""
+        detection = detect_crisis("This is so frustrating, I hate this")
+        assert detection.level in [CrisisLevel.NONE, CrisisLevel.LOW]
+    
+    def test_context_clues_increase_confidence(self):
+        """Multiple context clues should increase confidence."""
+        detection = detect_crisis("I'm so depressed and alone. No one cares about me.")
+        assert detection.detected
+        assert len(detection.triggers) > 0
+    
+    def test_escalating_distress_in_history(self):
+        """Escalating distress in conversation history should be detected."""
+        history = [
+            {"role": "user", "content": "I've been feeling depressed"},
+            {"role": "assistant", "content": "I'm sorry to hear that..."},
+            {"role": "user", "content": "Everything feels hopeless"},
+            {"role": "assistant", "content": "I'm here for you..."},
+            {"role": "user", "content": "I'm in so much pain"},
+        ]
+        detection = detect_crisis("I can't take it anymore", history)
+        assert detection.detected
+
+
+class TestCrisisResponse:
+    """Test crisis response generation."""
+    
+    def test_high_crisis_response(self):
+        """High crisis should include safety check and 988."""
+        response = get_crisis_response(CrisisLevel.HIGH)
+        assert "Are you safe right now?" in response["response"]
+        assert "988" in response["response"]
+        assert response["requires_human_notification"]
+    
+    def test_imminent_crisis_response(self):
+        """Imminent crisis should include 911."""
+        response = get_crisis_response(CrisisLevel.IMMINENT)
+        assert "911" in response["response"]
+        assert "988" in response["response"]
+        assert response["requires_human_notification"]
+    
+    def test_gospel_included(self):
+        """All crisis responses should include the gospel."""
+        for level in [CrisisLevel.LOW, CrisisLevel.HIGH, CrisisLevel.IMMINENT]:
+            response = get_crisis_response(level)
+            assert "Jesus" in response["response"]
+    
+    def test_low_crisis_no_immediate_danger(self):
+        """Low crisis should not mention immediate danger."""
+        response = get_crisis_response(CrisisLevel.LOW)
+        assert "immediate danger" not in response["response"].lower()
+        assert "911" not in response["response"]
+    
+    def test_lifeline_info_included(self):
+        """Response should include lifeline information."""
+        response = get_crisis_response(CrisisLevel.HIGH)
+        assert "lifeline" in response
+        assert "988" in response["lifeline"]["phone"]
+        assert "988lifeline.org" in response["lifeline"]["chat"]
+
+
+class TestToolBlocking:
+    """Test tool blocking during crisis."""
+    
+    def test_block_tools_on_high_crisis(self):
+        """Tools should be blocked during high crisis."""
+        detection = CrisisDetection(
+            detected=True,
+            level=CrisisLevel.HIGH,
+            confidence=0.9,
+            triggers=["test"],
+            requires_protocol=True,
+        )
+        assert should_block_tools(detection)
+    
+    def test_block_tools_on_imminent(self):
+        """Tools should be blocked during imminent crisis."""
+        detection = CrisisDetection(
+            detected=True,
+            level=CrisisLevel.IMMINENT,
+            confidence=0.95,
+            triggers=["test"],
+            requires_protocol=True,
+        )
+        assert should_block_tools(detection)
+    
+    def test_no_block_on_low_crisis(self):
+        """Tools should not be blocked for low crisis."""
+        detection = CrisisDetection(
+            detected=True,
+            level=CrisisLevel.LOW,
+            confidence=0.5,
+            triggers=["test"],
+            requires_protocol=True,
+        )
+        assert not should_block_tools(detection)
+    
+    def test_no_block_when_no_crisis(self):
+        """Tools should not be blocked when no crisis."""
+        detection = CrisisDetection(
+            detected=False,
+            level=CrisisLevel.NONE,
+            confidence=0.0,
+            triggers=[],
+            requires_protocol=False,
+        )
+        assert not should_block_tools(detection)
+
+
+class TestHumanNotification:
+    """Test human notification generation."""
+    
+    def test_notification_includes_level(self):
+        """Notification should include crisis level."""
+        detection = CrisisDetection(
+            detected=True,
+            level=CrisisLevel.HIGH,
+            confidence=0.9,
+            triggers=["pattern: test"],
+            requires_protocol=True,
+        )
+        notification = get_human_notification(detection, "test message")
+        assert notification["level"] == "high"
+        assert notification["action_required"]
+    
+    def test_notification_includes_preview(self):
+        """Notification should include message preview."""
+        detection = CrisisDetection(
+            detected=True,
+            level=CrisisLevel.HIGH,
+            confidence=0.9,
+            triggers=[],
+            requires_protocol=True,
+        )
+        long_message = "x" * 500
+        notification = get_human_notification(detection, long_message)
+        assert len(notification["user_message_preview"]) <= 200
+
+
+class TestCrisisProtocol:
+    """Test the CrisisProtocol class."""
+    
+    def test_protocol_check_message(self):
+        """Protocol should detect crisis."""
+        protocol = CrisisProtocol()
+        detection = protocol.check_message("I want to die")
+        assert detection.detected
+        assert protocol.detection_count == 1
+    
+    def test_protocol_get_response(self):
+        """Protocol should return crisis response."""
+        protocol = CrisisProtocol()
+        detection = protocol.check_message("I want to die")
+        response = protocol.get_response(detection)
+        assert "Are you safe" in response["response"]
+    
+    def test_protocol_blocks_tools(self):
+        """Protocol should block tools during crisis."""
+        protocol = CrisisProtocol()
+        detection = protocol.check_message("I want to die")
+        assert protocol.should_block_tools(detection)
+    
+    def test_protocol_no_block_normal(self):
+        """Protocol should not block tools for normal messages."""
+        protocol = CrisisProtocol()
+        detection = protocol.check_message("Hello, how are you?")
+        assert not protocol.should_block_tools(detection)
+
+
+if __name__ == "__main__":
+    pytest.main([__file__, "-v"])