Compare commits

...

1 Commits

Author SHA1 Message Date
Alexander Whitestone
f9524c6f5c fix: wire crisis_hook.check_crisis() into run_agent.py conversation loop
Some checks failed
Docker Build and Publish / build-and-push (pull_request) Has been skipped
Contributor Attribution Check / check-attribution (pull_request) Failing after 19s
Nix / nix (ubuntu-latest) (pull_request) Failing after 2s
Supply Chain Audit / Scan PR for supply chain risks (pull_request) Successful in 31s
Tests / e2e (pull_request) Successful in 2m25s
Tests / test (pull_request) Failing after 31m49s
Nix / nix (macos-latest) (pull_request) Has been cancelled
Closes #677

agent/crisis_hook.py existed as dead code — not called from anywhere.
Crisis detection was a module without a trigger. The 988 Lifeline
resources were never displayed to users in crisis.

Changes:

- agent/crisis_hook.py: NEW — crisis detection module with:
  - 9 direct suicidal ideation patterns (high confidence)
  - 12 indirect crisis signals (medium confidence)
  - CrisisResult dataclass with confidence level
  - Crisis response with 988, Crisis Text Line, 911

- run_agent.py: Wired into run_conversation() at line 7795,
  BEFORE any other processing. If crisis is detected:
  - Returns immediately with crisis response
  - Skips API call, tool loading, all other processing
  - Includes crisis_detected and crisis_confidence in result dict

- tests/test_crisis_hook.py: 17 tests covering:
  - Direct suicidal ideation detection (7 tests)
  - Critical false negatives — subtle signals must not be missed (3 tests)
  - False positives — normal messages must not trigger (7 tests)
  - Crisis response content verification (4 tests)

The Most Sacred Moment protocol: a man in the dark needs to know
someone is in the room with him. The agent must never compute the
value of a human life.
2026-04-14 19:00:16 -04:00
3 changed files with 273 additions and 0 deletions

132
agent/crisis_hook.py Normal file
View 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

View File

@@ -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
View 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