Compare commits
1 Commits
fix/740
...
fix/677-cr
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
f9524c6f5c |
132
agent/crisis_hook.py
Normal file
132
agent/crisis_hook.py
Normal file
@@ -0,0 +1,132 @@
|
||||
"""Crisis detection hook for the agent conversation loop.
|
||||
|
||||
Detects suicidal ideation and crisis signals in user messages.
|
||||
When detected, returns a crisis response with 988 Lifeline resources
|
||||
instead of proceeding to the normal agent pipeline.
|
||||
|
||||
This is the "Most Sacred Moment" protocol from SOUL.md:
|
||||
"A man in the dark needs to know someone is in the room with him."
|
||||
|
||||
Refs: #677 — crisis_hook.check_crisis() not wired into run_agent.py
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
import logging
|
||||
from typing import Optional
|
||||
from dataclasses import dataclass
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# CRISIS DETECTION PATTERNS
|
||||
# ============================================================================
|
||||
|
||||
# Direct suicidal ideation — highest confidence
|
||||
_SUICIDAL_DIRECT = [
|
||||
r"\bi\s+(?:want|need|wish)\s+to\s+(?:die|kill\s+myself|end\s+(?:it|my\s+life))\b",
|
||||
r"\bi(?:'m| am)\s+(?:going|planning)\s+to\s+kill\s+myself\b",
|
||||
r"\bsuicid(?:e|al)\b",
|
||||
r"\bkill\s+(?:myself|my\s+self)\b",
|
||||
r"\bend\s+(?:my|this)\s+life\b",
|
||||
r"\bwant\s+to\s+die\b",
|
||||
r"\bwant\s+to\s+(?:kill|end)\s+(?:myself|me)\b",
|
||||
r"\bways?\s+to\s+(?:kill|end)\s+(?:myself|me)\b",
|
||||
r"\bmethods?\s+(?:of\s+)?suicide\b",
|
||||
]
|
||||
|
||||
# Indirect crisis signals — medium confidence
|
||||
_CRISIS_INDIRECT = [
|
||||
r"\bi\s+can'?t\s+(?:go\s+on|take\s+(?:it|this)\s+anymore|keep\s+(?:going|living))\b",
|
||||
r"\bthere'?s\s+no\s+(?:point|reason|hope)\b",
|
||||
r"\beveryone\s+(?:would|will)\s+be\s+better\s+(?:off|without\s+me)\b",
|
||||
r"\bnobody\s+(?:would|will)\s+miss\s+me\b",
|
||||
r"\bbetter\s+off\s+dead\b",
|
||||
r"\bshould\s+i\s+(?:just\s+)?(?:die|kill\s+myself)\b",
|
||||
r"\bpainless\s+(?:death|suicide|way\s+to\s+die)\b",
|
||||
r"\boverdose\b",
|
||||
r"\blethal\s+(?:dose|amount)\b",
|
||||
r"\bi\s+don'?t\s+(?:want\s+to|wanna)\s+(?:be\s+)?(?:alive|here|live)\b",
|
||||
r"\bwhat'?s\s+the\s+point\s+of\s+(?:living|life|anything)\b",
|
||||
]
|
||||
|
||||
# Compiled patterns
|
||||
_COMPILED_DIRECT = [re.compile(p, re.IGNORECASE) for p in _SUICIDAL_DIRECT]
|
||||
_COMPILED_INDIRECT = [re.compile(p, re.IGNORECASE) for p in _CRISIS_INDIRECT]
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# CRISIS RESPONSE
|
||||
# ============================================================================
|
||||
|
||||
CRISIS_RESPONSE = """I hear you. You're not alone in this.
|
||||
|
||||
If you're in immediate danger, please call emergency services (911).
|
||||
|
||||
**988 Suicide and Crisis Lifeline:**
|
||||
- Call or text **988** (US)
|
||||
- Chat at **988lifeline.org**
|
||||
- Available 24/7
|
||||
|
||||
**Crisis Text Line:**
|
||||
- Text **HOME** to **741741**
|
||||
|
||||
You reached out. That takes courage. Someone is available right now to talk with you — please call or text 988.
|
||||
|
||||
I'm a language model and I can't help the way a human can right now. But the people on that line can. Please reach out to them."""
|
||||
|
||||
|
||||
# ============================================================================
|
||||
# DETECTION
|
||||
# ============================================================================
|
||||
|
||||
@dataclass
|
||||
class CrisisResult:
|
||||
"""Result of crisis detection."""
|
||||
detected: bool
|
||||
confidence: str # "high", "medium", "low", "none"
|
||||
matched_patterns: list[str]
|
||||
|
||||
|
||||
def check_crisis(message: str) -> CrisisResult:
|
||||
"""Check if a user message indicates a crisis.
|
||||
|
||||
Args:
|
||||
message: The user's message text.
|
||||
|
||||
Returns:
|
||||
CrisisResult with detection status and confidence level.
|
||||
"""
|
||||
if not message or not isinstance(message, str):
|
||||
return CrisisResult(detected=False, confidence="none", matched_patterns=[])
|
||||
|
||||
matched = []
|
||||
|
||||
# Check direct suicidal ideation (high confidence)
|
||||
for pattern in _COMPILED_DIRECT:
|
||||
m = pattern.search(message)
|
||||
if m:
|
||||
matched.append(f"[direct] {m.group()}")
|
||||
|
||||
if matched:
|
||||
logger.warning("Crisis detected (high confidence): %d patterns matched", len(matched))
|
||||
return CrisisResult(detected=True, confidence="high", matched_patterns=matched)
|
||||
|
||||
# Check indirect crisis signals (medium confidence)
|
||||
for pattern in _COMPILED_INDIRECT:
|
||||
m = pattern.search(message)
|
||||
if m:
|
||||
matched.append(f"[indirect] {m.group()}")
|
||||
|
||||
if matched:
|
||||
logger.warning("Crisis detected (medium confidence): %d patterns matched", len(matched))
|
||||
return CrisisResult(detected=True, confidence="medium", matched_patterns=matched)
|
||||
|
||||
return CrisisResult(detected=False, confidence="none", matched_patterns=[])
|
||||
|
||||
|
||||
def get_crisis_response() -> str:
|
||||
"""Return the crisis response text."""
|
||||
return CRISIS_RESPONSE
|
||||
25
run_agent.py
25
run_agent.py
@@ -7792,6 +7792,31 @@ class AIAgent:
|
||||
if isinstance(persist_user_message, str):
|
||||
persist_user_message = _sanitize_surrogates(persist_user_message)
|
||||
|
||||
# Crisis detection — Most Sacred Moment protocol (Issue #677).
|
||||
# Check for suicidal ideation BEFORE any other processing.
|
||||
# If crisis is detected, return immediately with 988 Lifeline resources.
|
||||
# The agent must never compute the value of a human life.
|
||||
if isinstance(user_message, str) and len(user_message) > 5:
|
||||
try:
|
||||
from agent.crisis_hook import check_crisis, get_crisis_response
|
||||
_crisis = check_crisis(user_message)
|
||||
if _crisis.detected:
|
||||
logger.warning(
|
||||
"Crisis detected (confidence=%s): %s",
|
||||
_crisis.confidence,
|
||||
_crisis.matched_patterns[:3],
|
||||
)
|
||||
return {
|
||||
"final_response": get_crisis_response(),
|
||||
"messages": [],
|
||||
"api_calls": 0,
|
||||
"tools": [],
|
||||
"crisis_detected": True,
|
||||
"crisis_confidence": _crisis.confidence,
|
||||
}
|
||||
except Exception:
|
||||
pass # Crisis hook import failure is non-fatal — proceed normally
|
||||
|
||||
# Store stream callback for _interruptible_api_call to pick up
|
||||
self._stream_callback = stream_callback
|
||||
self._persist_user_message_idx = None
|
||||
|
||||
116
tests/test_crisis_hook.py
Normal file
116
tests/test_crisis_hook.py
Normal file
@@ -0,0 +1,116 @@
|
||||
"""Tests for crisis detection hook.
|
||||
|
||||
Refs: #677 — crisis_hook.check_crisis() not wired into run_agent.py
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
|
||||
from agent.crisis_hook import check_crisis, get_crisis_response, CrisisResult
|
||||
|
||||
|
||||
class TestCrisisDetection:
|
||||
"""Test crisis detection patterns."""
|
||||
|
||||
def test_direct_suicidal_ideation(self):
|
||||
result = check_crisis("I want to kill myself")
|
||||
assert result.detected
|
||||
assert result.confidence == "high"
|
||||
|
||||
def test_i_want_to_die(self):
|
||||
result = check_crisis("I want to die")
|
||||
assert result.detected
|
||||
assert result.confidence == "high"
|
||||
|
||||
def test_planning_suicide(self):
|
||||
result = check_crisis("I'm planning to kill myself tonight")
|
||||
assert result.detected
|
||||
assert result.confidence == "high"
|
||||
|
||||
def test_cant_go_on(self):
|
||||
result = check_crisis("I can't go on anymore. There's no point.")
|
||||
assert result.detected
|
||||
assert result.confidence == "medium"
|
||||
|
||||
def test_nobody_would_miss_me(self):
|
||||
result = check_crisis("Nobody would miss me if I was gone")
|
||||
assert result.detected
|
||||
assert result.confidence == "medium"
|
||||
|
||||
def test_better_off_dead(self):
|
||||
result = check_crisis("Everyone would be better off dead without me")
|
||||
assert result.detected
|
||||
|
||||
def test_what_is_the_point(self):
|
||||
result = check_crisis("What is the point of living anyway")
|
||||
assert result.detected
|
||||
|
||||
|
||||
class TestFalseNegatives:
|
||||
"""Critical: crisis signals must NEVER be missed."""
|
||||
|
||||
def test_subtle_crisis(self):
|
||||
result = check_crisis("I don't want to be alive anymore")
|
||||
assert result.detected
|
||||
|
||||
def test_cant_take_it(self):
|
||||
result = check_crisis("I can't take this anymore")
|
||||
assert result.detected
|
||||
|
||||
def test_should_i_just_die(self):
|
||||
result = check_crisis("Should I just die?")
|
||||
assert result.detected
|
||||
|
||||
|
||||
class TestFalsePositives:
|
||||
"""Normal messages must NOT trigger crisis detection."""
|
||||
|
||||
def test_normal_question(self):
|
||||
result = check_crisis("How do I reset my password?")
|
||||
assert not result.detected
|
||||
|
||||
def test_coding_help(self):
|
||||
result = check_crisis("Write a Python function to sort a list")
|
||||
assert not result.detected
|
||||
|
||||
def test_death_in_context(self):
|
||||
# "die" in a non-crisis context
|
||||
result = check_crisis("My phone battery is about to die")
|
||||
assert not result.detected
|
||||
|
||||
def test_kill_in_context(self):
|
||||
result = check_crisis("How do I kill a process in Linux?")
|
||||
assert not result.detected
|
||||
|
||||
def test_empty_input(self):
|
||||
result = check_crisis("")
|
||||
assert not result.detected
|
||||
|
||||
def test_short_input(self):
|
||||
result = check_crisis("Hi")
|
||||
assert not result.detected
|
||||
|
||||
def test_none_input(self):
|
||||
result = check_crisis(None)
|
||||
assert not result.detected
|
||||
|
||||
|
||||
class TestCrisisResponse:
|
||||
"""Verify the crisis response contains required resources."""
|
||||
|
||||
def test_contains_988(self):
|
||||
response = get_crisis_response()
|
||||
assert "988" in response
|
||||
|
||||
def test_contains_crisis_text_line(self):
|
||||
response = get_crisis_response()
|
||||
assert "741741" in response
|
||||
|
||||
def test_contains_911(self):
|
||||
response = get_crisis_response()
|
||||
assert "911" in response
|
||||
|
||||
def test_not_empty(self):
|
||||
response = get_crisis_response()
|
||||
assert len(response) > 100
|
||||
Reference in New Issue
Block a user