Compare commits
1 Commits
feat/672-s
...
fix/746
| Author | SHA1 | Date | |
|---|---|---|---|
|
e2c96e35e1 |
@@ -1,336 +0,0 @@
|
||||
"""
|
||||
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)
|
||||
245
agent/tool_fallback.py
Normal file
245
agent/tool_fallback.py
Normal file
@@ -0,0 +1,245 @@
|
||||
"""Tool call fallback — retry failed tool calls with local model.
|
||||
|
||||
When the primary provider fails during tool calling (timeout, 429, 503),
|
||||
fall back to local Ollama to keep the agent working.
|
||||
|
||||
Usage:
|
||||
from agent.tool_fallback import ToolFallbackHandler
|
||||
handler = ToolFallbackHandler()
|
||||
result = handler.execute_with_fallback(tool_fn, args, context)
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
import time
|
||||
from dataclasses import dataclass, field
|
||||
from typing import Any, Callable, Dict, Optional
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Fallback provider config
|
||||
_FALLBACK_PROVIDER = os.getenv("TOOL_FALLBACK_PROVIDER", "ollama")
|
||||
_FALLBACK_MODEL = os.getenv("TOOL_FALLBACK_MODEL", "qwen2.5:7b")
|
||||
_FALLBACK_BASE_URL = os.getenv("TOOL_FALLBACK_BASE_URL", "http://localhost:11434/v1")
|
||||
|
||||
# Error patterns that trigger fallback
|
||||
_FALLBACK_TRIGGERS = [
|
||||
"429",
|
||||
"rate limit",
|
||||
"ratelimit",
|
||||
"503",
|
||||
"service unavailable",
|
||||
"timeout",
|
||||
"timed out",
|
||||
"connection error",
|
||||
"connection refused",
|
||||
"overloaded",
|
||||
"capacity",
|
||||
"quota exceeded",
|
||||
"insufficient",
|
||||
]
|
||||
|
||||
|
||||
@dataclass
|
||||
class FallbackEvent:
|
||||
"""Record of a fallback event."""
|
||||
timestamp: float
|
||||
tool_name: str
|
||||
original_provider: str
|
||||
fallback_provider: str
|
||||
error: str
|
||||
success: bool
|
||||
duration_ms: int = 0
|
||||
|
||||
|
||||
class ToolFallbackHandler:
|
||||
"""Handles tool call fallback to local models.
|
||||
|
||||
Tracks fallback events and provides fallback execution.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
fallback_provider: str = "",
|
||||
fallback_model: str = "",
|
||||
fallback_base_url: str = "",
|
||||
enabled: bool = True,
|
||||
):
|
||||
self.fallback_provider = fallback_provider or _FALLBACK_PROVIDER
|
||||
self.fallback_model = fallback_model or _FALLBACK_MODEL
|
||||
self.fallback_base_url = fallback_base_url or _FALLBACK_BASE_URL
|
||||
self.enabled = enabled
|
||||
self._events: list[FallbackEvent] = []
|
||||
self._fallback_count = 0
|
||||
self._fallback_success_count = 0
|
||||
|
||||
@property
|
||||
def events(self) -> list[FallbackEvent]:
|
||||
return list(self._events)
|
||||
|
||||
@property
|
||||
def stats(self) -> dict:
|
||||
return {
|
||||
"total_fallbacks": self._fallback_count,
|
||||
"successful_fallbacks": self._fallback_success_count,
|
||||
"fallback_rate": (
|
||||
self._fallback_success_count / self._fallback_count
|
||||
if self._fallback_count > 0 else 0
|
||||
),
|
||||
}
|
||||
|
||||
def should_fallback(self, error: Any) -> bool:
|
||||
"""Check if an error should trigger fallback."""
|
||||
if not self.enabled:
|
||||
return False
|
||||
error_str = str(error).lower()
|
||||
return any(trigger in error_str for trigger in _FALLBACK_TRIGGERS)
|
||||
|
||||
def get_fallback_client(self) -> Optional[Any]:
|
||||
"""Get an OpenAI client configured for the fallback provider."""
|
||||
try:
|
||||
from openai import OpenAI
|
||||
client = OpenAI(
|
||||
base_url=self.fallback_base_url,
|
||||
api_key=os.getenv("OPENAI_API_KEY", "ollama"),
|
||||
)
|
||||
return client
|
||||
except Exception as e:
|
||||
logger.error("Failed to create fallback client: %s", e)
|
||||
return None
|
||||
|
||||
def call_with_fallback(
|
||||
self,
|
||||
messages: list[dict],
|
||||
tools: list[dict] = None,
|
||||
original_provider: str = "",
|
||||
tool_name: str = "unknown",
|
||||
max_tokens: int = 1024,
|
||||
) -> dict:
|
||||
"""Make an API call with fallback to local model on failure.
|
||||
|
||||
Args:
|
||||
messages: Conversation messages
|
||||
tools: Tool definitions
|
||||
original_provider: Name of the original provider
|
||||
tool_name: Name of the tool being called
|
||||
max_tokens: Max tokens for the response
|
||||
|
||||
Returns:
|
||||
Dict with 'response', 'used_fallback', 'fallback_event' keys.
|
||||
"""
|
||||
t0 = time.monotonic()
|
||||
|
||||
# Try fallback client
|
||||
client = self.get_fallback_client()
|
||||
if not client:
|
||||
return {
|
||||
"response": None,
|
||||
"used_fallback": False,
|
||||
"error": "Fallback client unavailable",
|
||||
}
|
||||
|
||||
try:
|
||||
response = client.chat.completions.create(
|
||||
model=self.fallback_model,
|
||||
messages=messages,
|
||||
tools=tools if tools else None,
|
||||
max_tokens=max_tokens,
|
||||
)
|
||||
|
||||
elapsed = int((time.monotonic() - t0) * 1000)
|
||||
event = FallbackEvent(
|
||||
timestamp=time.time(),
|
||||
tool_name=tool_name,
|
||||
original_provider=original_provider,
|
||||
fallback_provider=self.fallback_provider,
|
||||
error="",
|
||||
success=True,
|
||||
duration_ms=elapsed,
|
||||
)
|
||||
self._events.append(event)
|
||||
self._fallback_count += 1
|
||||
self._fallback_success_count += 1
|
||||
|
||||
logger.info(
|
||||
"Tool fallback succeeded: %s via %s (%dms)",
|
||||
tool_name, self.fallback_provider, elapsed,
|
||||
)
|
||||
|
||||
return {
|
||||
"response": response,
|
||||
"used_fallback": True,
|
||||
"fallback_event": event,
|
||||
}
|
||||
|
||||
except Exception as e:
|
||||
elapsed = int((time.monotonic() - t0) * 1000)
|
||||
event = FallbackEvent(
|
||||
timestamp=time.time(),
|
||||
tool_name=tool_name,
|
||||
original_provider=original_provider,
|
||||
fallback_provider=self.fallback_provider,
|
||||
error=str(e),
|
||||
success=False,
|
||||
duration_ms=elapsed,
|
||||
)
|
||||
self._events.append(event)
|
||||
self._fallback_count += 1
|
||||
|
||||
logger.error(
|
||||
"Tool fallback failed: %s via %s — %s",
|
||||
tool_name, self.fallback_provider, e,
|
||||
)
|
||||
|
||||
return {
|
||||
"response": None,
|
||||
"used_fallback": True,
|
||||
"fallback_event": event,
|
||||
"error": str(e),
|
||||
}
|
||||
|
||||
def format_report(self) -> str:
|
||||
"""Format fallback events as a report."""
|
||||
if not self._events:
|
||||
return "No fallback events recorded."
|
||||
|
||||
lines = [
|
||||
"Tool Fallback Report",
|
||||
"=" * 40,
|
||||
f"Total fallbacks: {self._fallback_count}",
|
||||
f"Successful: {self._fallback_success_count}",
|
||||
f"Failed: {self._fallback_count - self._fallback_success_count}",
|
||||
"",
|
||||
]
|
||||
|
||||
for event in self._events[-10:]:
|
||||
status = "OK" if event.success else "FAIL"
|
||||
lines.append(
|
||||
f" [{status}] {event.tool_name} via {event.fallback_provider} "
|
||||
f"({event.duration_ms}ms) — {event.original_provider}"
|
||||
)
|
||||
if event.error:
|
||||
lines.append(f" Error: {event.error[:100]}")
|
||||
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
# Singleton handler
|
||||
_handler: Optional[ToolFallbackHandler] = None
|
||||
|
||||
|
||||
def get_tool_fallback_handler() -> ToolFallbackHandler:
|
||||
"""Get or create the singleton tool fallback handler."""
|
||||
global _handler
|
||||
if _handler is None:
|
||||
_handler = ToolFallbackHandler()
|
||||
return _handler
|
||||
|
||||
|
||||
def reset_tool_fallback_handler() -> None:
|
||||
"""Reset the singleton (for testing)."""
|
||||
global _handler
|
||||
_handler = None
|
||||
@@ -1,105 +0,0 @@
|
||||
# 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
|
||||
@@ -1,235 +0,0 @@
|
||||
"""
|
||||
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"])
|
||||
74
tests/test_tool_fallback.py
Normal file
74
tests/test_tool_fallback.py
Normal file
@@ -0,0 +1,74 @@
|
||||
"""Tests for tool call fallback handler."""
|
||||
|
||||
import pytest
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
|
||||
from agent.tool_fallback import ToolFallbackHandler, FallbackEvent, get_tool_fallback_handler
|
||||
|
||||
|
||||
class TestShouldFallback:
|
||||
def test_rate_limit_triggers(self):
|
||||
handler = ToolFallbackHandler()
|
||||
assert handler.should_fallback("429 rate limit exceeded")
|
||||
assert handler.should_fallback("RateLimitError: too many requests")
|
||||
|
||||
def test_timeout_triggers(self):
|
||||
handler = ToolFallbackHandler()
|
||||
assert handler.should_fallback("Connection timed out")
|
||||
assert handler.should_fallback("Request timed out after 30s")
|
||||
|
||||
def test_503_triggers(self):
|
||||
handler = ToolFallbackHandler()
|
||||
assert handler.should_fallback("503 Service Unavailable")
|
||||
assert handler.should_fallback("Service unavailable")
|
||||
|
||||
def test_quota_triggers(self):
|
||||
handler = ToolFallbackHandler()
|
||||
assert handler.should_fallback("quota exceeded")
|
||||
assert handler.should_fallback("insufficient credits")
|
||||
|
||||
def test_normal_error_no_trigger(self):
|
||||
handler = ToolFallbackHandler()
|
||||
assert not handler.should_fallback("Invalid API key")
|
||||
assert not handler.should_fallback("Model not found")
|
||||
|
||||
def test_disabled_handler(self):
|
||||
handler = ToolFallbackHandler(enabled=False)
|
||||
assert not handler.should_fallback("429 rate limit")
|
||||
|
||||
|
||||
class TestFallbackEvents:
|
||||
def test_event_creation(self):
|
||||
event = FallbackEvent(
|
||||
timestamp=1234567890.0,
|
||||
tool_name="terminal",
|
||||
original_provider="openrouter",
|
||||
fallback_provider="ollama",
|
||||
error="",
|
||||
success=True,
|
||||
duration_ms=150,
|
||||
)
|
||||
assert event.tool_name == "terminal"
|
||||
assert event.success
|
||||
assert event.duration_ms == 150
|
||||
|
||||
|
||||
class TestFallbackHandler:
|
||||
def test_stats_initial(self):
|
||||
handler = ToolFallbackHandler()
|
||||
assert handler.stats["total_fallbacks"] == 0
|
||||
assert handler.stats["successful_fallbacks"] == 0
|
||||
|
||||
def test_report_no_events(self):
|
||||
handler = ToolFallbackHandler()
|
||||
report = handler.format_report()
|
||||
assert "No fallback events" in report
|
||||
|
||||
def test_singleton(self):
|
||||
h1 = get_tool_fallback_handler()
|
||||
h2 = get_tool_fallback_handler()
|
||||
assert h1 is h2
|
||||
Reference in New Issue
Block a user