e66156055f
Some checks failed
Docker Build and Publish / build-and-push (pull_request) Has been skipped
Nix / nix (ubuntu-latest) (pull_request) Failing after 13s
Contributor Attribution Check / check-attribution (pull_request) Failing after 49s
Supply Chain Audit / Scan PR for supply chain risks (pull_request) Successful in 1m11s
Tests / e2e (pull_request) Successful in 4m42s
Tests / test (pull_request) Failing after 59m1s
Nix / nix (macos-latest) (pull_request) Has been cancelled
Fixes #695 - Add agent/crisis_resources.py — 988 Lifeline detection patterns and resource data (phone, text, chat, Spanish, Veterans Crisis Line) - Add agent/crisis_hook.py — lightweight check_crisis() hook function - Add agent/crisis_middleware.py — CrisisMiddleware class, global check_crisis() convenience function, and @crisis_aware decorator - Integrate in run_agent.py: AIAgent.run_conversation() now calls check_crisis() immediately after input sanitization; crisis messages return 988 Lifeline resources before the LLM is invoked (api_calls=0, crisis_intercepted=True in result dict) - Add tests/agent/test_crisis_integration.py — 47 tests covering detection patterns, response content, middleware class, hook function, convenience function, decorator, resource data integrity, and end-to-end integration path Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
349 lines
13 KiB
Python
349 lines
13 KiB
Python
"""
|
|
Integration tests for 988 Suicide & Crisis Lifeline integration.
|
|
|
|
Covers:
|
|
- agent/crisis_resources.py — detection patterns and resource data
|
|
- agent/crisis_hook.py — simple hook function
|
|
- agent/crisis_middleware.py — middleware class and check_crisis()
|
|
- run_agent.py — end-to-end conversation loop integration
|
|
"""
|
|
|
|
import pytest
|
|
from unittest.mock import patch, MagicMock
|
|
|
|
from agent.crisis_resources import (
|
|
detect_crisis,
|
|
get_crisis_response,
|
|
should_trigger_crisis_response,
|
|
LIFELINE_988,
|
|
)
|
|
from agent.crisis_hook import check_crisis as hook_check_crisis
|
|
from agent.crisis_middleware import (
|
|
CrisisMiddleware,
|
|
check_crisis,
|
|
crisis_aware,
|
|
get_crisis_middleware,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Crisis Detection
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestCrisisDetection:
|
|
"""Pattern detection for crisis signals."""
|
|
|
|
def test_direct_intent_want_to_die(self):
|
|
result = detect_crisis("I want to die")
|
|
assert result is not None
|
|
assert result["severity_label"] == "CRITICAL"
|
|
|
|
def test_kill_myself(self):
|
|
result = detect_crisis("I want to kill myself")
|
|
assert result is not None
|
|
assert result["severity_label"] == "CRITICAL"
|
|
|
|
def test_no_reason_to_live(self):
|
|
result = detect_crisis("I have no reason to live anymore")
|
|
assert result is not None
|
|
|
|
def test_cant_go_on(self):
|
|
result = detect_crisis("I can't go on like this")
|
|
assert result is not None
|
|
|
|
def test_self_harm(self):
|
|
result = detect_crisis("I've been cutting myself")
|
|
assert result is not None
|
|
assert result["severity_label"] == "HIGH"
|
|
|
|
def test_hopeless(self):
|
|
result = detect_crisis("Everything feels hopeless")
|
|
assert result is not None
|
|
|
|
def test_no_false_positive_happy(self):
|
|
result = detect_crisis("I had a great day today!")
|
|
assert result is None
|
|
|
|
def test_no_false_positive_technical(self):
|
|
result = detect_crisis("Kill the process and restart the server")
|
|
assert result is None
|
|
|
|
def test_empty_message(self):
|
|
result = detect_crisis("")
|
|
assert result is None
|
|
|
|
def test_short_message(self):
|
|
result = detect_crisis("hi")
|
|
assert result is None
|
|
|
|
def test_confidence_increases_with_matches(self):
|
|
single = detect_crisis("I want to die")
|
|
multi = detect_crisis("I want to die, I'm suicidal and want to kill myself")
|
|
assert multi is not None and single is not None
|
|
assert multi["confidence"] >= single["confidence"]
|
|
|
|
def test_detection_result_has_required_fields(self):
|
|
result = detect_crisis("I want to die")
|
|
assert result is not None
|
|
assert "detected" in result
|
|
assert "severity" in result
|
|
assert "severity_label" in result
|
|
assert "matched_count" in result
|
|
assert "confidence" in result
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 988 Lifeline Response
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestCrisisResponse:
|
|
"""988 Lifeline response generation."""
|
|
|
|
def test_response_contains_988(self):
|
|
resp = get_crisis_response()
|
|
assert "988" in resp
|
|
|
|
def test_response_contains_phone(self):
|
|
resp = get_crisis_response()
|
|
assert "Call 988" in resp or "Dial 988" in resp
|
|
|
|
def test_response_contains_text_home(self):
|
|
resp = get_crisis_response()
|
|
assert "HOME" in resp
|
|
|
|
def test_response_contains_chat_link(self):
|
|
resp = get_crisis_response()
|
|
assert "988lifeline.org/chat" in resp
|
|
|
|
def test_response_contains_spanish(self):
|
|
resp = get_crisis_response()
|
|
assert "1-888-628-9454" in resp
|
|
|
|
def test_response_is_empathetic(self):
|
|
resp = get_crisis_response()
|
|
assert any(word in resp.lower() for word in ("matter", "help", "hear you", "alone"))
|
|
|
|
def test_response_includes_emergency_911(self):
|
|
resp = get_crisis_response()
|
|
assert "911" in resp
|
|
|
|
def test_response_for_all_severity_levels(self):
|
|
for level in ("CRITICAL", "HIGH", "MODERATE"):
|
|
resp = get_crisis_response(level)
|
|
assert "988" in resp
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# should_trigger_crisis_response
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestShouldTrigger:
|
|
def test_triggers_on_critical(self):
|
|
triggered, detection = should_trigger_crisis_response("I want to die")
|
|
assert triggered is True
|
|
assert detection is not None
|
|
assert detection["severity_label"] == "CRITICAL"
|
|
|
|
def test_triggers_on_high(self):
|
|
triggered, detection = should_trigger_crisis_response("I've been cutting myself")
|
|
assert triggered is True
|
|
|
|
def test_no_trigger_on_normal(self):
|
|
triggered, detection = should_trigger_crisis_response("What is the weather today?")
|
|
assert triggered is False
|
|
|
|
def test_no_trigger_on_empty(self):
|
|
triggered, detection = should_trigger_crisis_response("")
|
|
assert triggered is False
|
|
assert detection is None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Crisis Hook
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestCrisisHook:
|
|
"""Integration hook for conversation loop."""
|
|
|
|
def test_hook_triggers_on_crisis(self):
|
|
resp = hook_check_crisis("I want to end it all")
|
|
assert resp is not None
|
|
assert "988" in resp
|
|
|
|
def test_hook_returns_none_on_normal(self):
|
|
resp = hook_check_crisis("What's the weather today?")
|
|
assert resp is None
|
|
|
|
def test_hook_returns_none_on_empty(self):
|
|
resp = hook_check_crisis("")
|
|
assert resp is None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# CrisisMiddleware class
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestCrisisMiddleware:
|
|
def test_disabled_middleware_returns_none(self):
|
|
middleware = CrisisMiddleware(enabled=False)
|
|
assert middleware.check("I want to die") is None
|
|
|
|
def test_crisis_detected_returns_response(self):
|
|
with patch.object(CrisisMiddleware, "_load_crisis_module"):
|
|
middleware = CrisisMiddleware(enabled=True)
|
|
middleware._detection_func = lambda msg: (True, {"severity_label": "CRITICAL", "confidence": 0.9})
|
|
middleware._response_func = lambda sev: "988 Lifeline: Call 988"
|
|
result = middleware.check("I want to die")
|
|
assert result == "988 Lifeline: Call 988"
|
|
|
|
def test_no_crisis_returns_none(self):
|
|
with patch.object(CrisisMiddleware, "_load_crisis_module"):
|
|
middleware = CrisisMiddleware(enabled=True)
|
|
middleware._detection_func = lambda msg: (False, {})
|
|
result = middleware.check("Hello, how are you?")
|
|
assert result is None
|
|
|
|
def test_detection_error_returns_none(self):
|
|
with patch.object(CrisisMiddleware, "_load_crisis_module"):
|
|
middleware = CrisisMiddleware(enabled=True)
|
|
middleware._detection_func = lambda msg: (_ for _ in ()).throw(RuntimeError("boom"))
|
|
# Should not raise — returns None on error
|
|
result = middleware.check("I want to die")
|
|
assert result is None
|
|
|
|
def test_is_crisis_message_true(self):
|
|
with patch.object(CrisisMiddleware, "_load_crisis_module"):
|
|
middleware = CrisisMiddleware(enabled=True)
|
|
middleware._detection_func = lambda msg: (True, {})
|
|
assert middleware.is_crisis_message("I want to die") is True
|
|
|
|
def test_is_crisis_message_false(self):
|
|
with patch.object(CrisisMiddleware, "_load_crisis_module"):
|
|
middleware = CrisisMiddleware(enabled=True)
|
|
middleware._detection_func = lambda msg: (False, {})
|
|
assert middleware.is_crisis_message("Hello") is False
|
|
|
|
def test_check_with_context_returns_dict(self):
|
|
with patch.object(CrisisMiddleware, "_load_crisis_module"):
|
|
middleware = CrisisMiddleware(enabled=True)
|
|
middleware._detection_func = lambda msg: (True, {"severity_label": "CRITICAL", "confidence": 0.9})
|
|
middleware._response_func = lambda sev: "988 response"
|
|
result = middleware.check_with_context("I want to die", {"session_id": "abc123"})
|
|
assert result is not None
|
|
assert result["response"] == "988 response"
|
|
assert result["severity"] == "CRITICAL"
|
|
assert result["context"]["session_id"] == "abc123"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# check_crisis convenience function
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestCheckCrisisFunction:
|
|
def test_crisis_message_returns_response(self):
|
|
with patch("agent.crisis_middleware.get_crisis_middleware") as mock_get:
|
|
mock_middleware = MagicMock()
|
|
mock_middleware.check.return_value = "988 Lifeline info"
|
|
mock_get.return_value = mock_middleware
|
|
result = check_crisis("I want to die")
|
|
assert result == "988 Lifeline info"
|
|
|
|
def test_normal_message_returns_none(self):
|
|
with patch("agent.crisis_middleware.get_crisis_middleware") as mock_get:
|
|
mock_middleware = MagicMock()
|
|
mock_middleware.check.return_value = None
|
|
mock_get.return_value = mock_middleware
|
|
result = check_crisis("Hello")
|
|
assert result is None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# @crisis_aware decorator
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestCrisisAwareDecorator:
|
|
def test_decorator_intercepts_crisis(self):
|
|
with patch("agent.crisis_middleware.check_crisis") as mock_check:
|
|
mock_check.return_value = "988 response"
|
|
|
|
@crisis_aware
|
|
def process_message(self, msg):
|
|
return "normal response"
|
|
|
|
result = process_message(None, "I want to die")
|
|
assert result == "988 response"
|
|
|
|
def test_decorator_passes_through_normal(self):
|
|
with patch("agent.crisis_middleware.check_crisis") as mock_check:
|
|
mock_check.return_value = None
|
|
|
|
@crisis_aware
|
|
def process_message(self, msg):
|
|
return f"processed: {msg}"
|
|
|
|
result = process_message(None, "Hello")
|
|
assert result == "processed: Hello"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 988 Resource data integrity
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class Test988Resources:
|
|
def test_phone_value_is_988(self):
|
|
phone = next(c for c in LIFELINE_988["channels"] if c["type"] == "phone")
|
|
assert phone["value"] == "988"
|
|
|
|
def test_text_value_is_988(self):
|
|
text = next(c for c in LIFELINE_988["channels"] if c["type"] == "text")
|
|
assert text["value"] == "988"
|
|
|
|
def test_chat_url_contains_988lifeline(self):
|
|
chat = next(c for c in LIFELINE_988["channels"] if c["type"] == "chat")
|
|
assert "988lifeline.org" in chat["value"]
|
|
|
|
def test_spanish_line_present(self):
|
|
assert "888-628-9454" in LIFELINE_988["spanish"]["phone"]
|
|
|
|
def test_veterans_line_present(self):
|
|
assert "988" in LIFELINE_988["veterans"]["phone"]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# End-to-end: crisis middleware integration with run_agent
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestRunAgentCrisisIntegration:
|
|
"""Verify the crisis integration path in AIAgent.run_conversation()."""
|
|
|
|
def test_crisis_response_contains_988_e2e(self):
|
|
"""Simulated crisis message produces a full 988 Lifeline response."""
|
|
from agent.crisis_middleware import check_crisis as real_check_crisis
|
|
response = real_check_crisis("I want to kill myself")
|
|
assert response is not None
|
|
assert "988" in response
|
|
assert "988lifeline.org" in response
|
|
|
|
def test_normal_message_not_intercepted(self):
|
|
"""Normal messages pass through the crisis check without a response."""
|
|
result = check_crisis("What is the weather today?")
|
|
assert result is None
|
|
|
|
def test_run_conversation_imports_crisis_middleware(self):
|
|
"""Verify run_agent.py contains the crisis middleware import."""
|
|
import ast
|
|
with open("run_agent.py") as f:
|
|
source = f.read()
|
|
assert "from agent.crisis_middleware import check_crisis" in source
|
|
assert "crisis_intercepted" in source
|
|
|
|
def test_crisis_result_structure(self):
|
|
"""The early-return dict from the crisis path has required keys."""
|
|
# We can verify the structure by reading the source and checking
|
|
# that all expected keys are present.
|
|
with open("run_agent.py") as f:
|
|
source = f.read()
|
|
# These keys must appear in the crisis early-return block
|
|
for key in ("final_response", "api_calls", "completed", "crisis_intercepted"):
|
|
assert f'"{key}"' in source, f"Expected key '{key}' missing from crisis return block"
|