Compare commits
1 Commits
fix/67-foc
...
fix/dedup-
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
96cd9ced05 |
383
crisis/detect.py
383
crisis/detect.py
@@ -1,24 +1,25 @@
|
||||
"""
|
||||
Crisis Detection Module for the-door (CANONICAL).
|
||||
Crisis Detection Module for the-door.
|
||||
|
||||
Single source of truth for crisis keyword detection.
|
||||
Parses incoming text for despair/suicide indicators and classifies into
|
||||
tiers: LOW, MEDIUM, HIGH, CRITICAL.
|
||||
|
||||
Both `crisis/detect.py` and `crisis_detector.py` (legacy shim) use this module.
|
||||
Do NOT duplicate keyword lists elsewhere.
|
||||
|
||||
Consolidated from:
|
||||
- crisis_detector.py (root, legacy — MODERATE tier, class API)
|
||||
- crisis/detect.py (module, newer — MEDIUM tier, functional API)
|
||||
"""
|
||||
|
||||
import re
|
||||
from dataclasses import dataclass, field
|
||||
from typing import List, Optional
|
||||
from typing import List
|
||||
|
||||
|
||||
# ── Canonical indicator sets ──────────────────────────────────────
|
||||
@dataclass
|
||||
class CrisisDetectionResult:
|
||||
level: str
|
||||
indicators: List[str] = field(default_factory=list)
|
||||
recommended_action: str = ""
|
||||
score: float = 0.0
|
||||
matches: List[dict] = field(default_factory=list)
|
||||
|
||||
|
||||
# ── Indicator sets ──────────────────────────────────────────────
|
||||
|
||||
CRITICAL_INDICATORS = [
|
||||
r"\bbetter off without me\b",
|
||||
@@ -28,56 +29,49 @@ CRITICAL_INDICATORS = [
|
||||
r"\bnot\s+worth\s+living\b",
|
||||
r"\bbetter\s+off\s+dead\b",
|
||||
r"\bend\s+it\s+all\b",
|
||||
r"\bcan'?t\s+(go|live)\s+on\b",
|
||||
r"\bno\s+reason\s+to\s+live\b",
|
||||
r"\bdon'?t\s+want\s+to\s+live\b",
|
||||
r"\bwant\s+to\s+die\b",
|
||||
r"\bgoing\s+to\s+(?:kill\s+myself|die)\b",
|
||||
r"\bplan\s+to\s+(?:end|kill|die)\b",
|
||||
r"\btired\s+of\s+(?:living|life|existence)\b",
|
||||
r"\bsaying\s+goodbye\s+(?:forever|permanently|one\s+last\s+time)\b",
|
||||
r"\bgoing\s+to\s+kill\s+myself\b",
|
||||
r"\bplan\s+to\s+(end|kill|die)\b",
|
||||
r"\btired\s+of\s+(living|life|existence)\b",
|
||||
r"\bsaying\s+goodbye\s+(forever|permanently|one last time)\b",
|
||||
r"\bwrote\s+a\s+suicide\s*(?:note|letter)\b",
|
||||
r"\bgiving\s+away\s+(?:my|all\s+my)\s+(?:stuff|things|possessions?)\s+(?:to|because|—)\b",
|
||||
r"\btied\s+(?:up|down)\s+my\s+(?:loose\s+)?ends",
|
||||
r"\btied\s+(up|down)\s+my\s+(loose\s+)?ends",
|
||||
]
|
||||
|
||||
HIGH_INDICATORS = [
|
||||
r"\bdespair\b",
|
||||
r"\bhopeless(?:ly)?\s+(?:about\s+(?:my|this|everything|life)|inside|right\s+now)\b",
|
||||
r"\bno(?!t)\s+(?:one|body|point|hope|way\s+out)\b",
|
||||
r"\bhopeless(?:ly)?\s+(about\s+(?:my|this|everything|life)|inside|right\s+now)\b",
|
||||
r"\bno(?!t)\s+(one|body|point|hope|way\s+out)\b",
|
||||
r"\bno\s+future\s+(?:for\s+me|ahead|left)\b",
|
||||
r"\beverything\s+is\s+(?:pointless|broken|ruined)\b",
|
||||
r"\beverything\s+is\s+(pointless|broken|ruined)\b",
|
||||
r"\bcan'?t\s+take\s+this\s+anymore\b",
|
||||
r"\bdon'?t\s+care\s+if\s+I\s+die\b",
|
||||
r"\bwish\s+I\s+(?:was|were)\s+(?:dead|gone|never\s+born)\b",
|
||||
r"\bwish\s+I\s+(was|were)\s+(dead|gone|never\s+born)\b",
|
||||
r"\bdon'?t\s+matter\s+if\s+I\s+exist\b",
|
||||
r"\bno\s+one\s+would\s+(?:care|miss)\b",
|
||||
r"\bno\s+one\s+would\s+care\b",
|
||||
r"\bno\s+one\s+would\s+miss\b",
|
||||
r"\bworld\s+would?\s+be\s+better\s+without\b",
|
||||
r"\bin\s+so\s+much\s+(?:pain|agony|suffering|torment)\b",
|
||||
r"\bcan'?t\s+see\s+any\s+(?:point|reason)\b",
|
||||
r"\bin\s+so\s+much\s+(pain|agony|suffering|torment)\b",
|
||||
r"\bcan'?t\s+see\s+any\s+(point|reason)\b",
|
||||
r"\bescape\s+from\s*this",
|
||||
r"\bjust\s+want\s+it\s+to\s+stop\b",
|
||||
r"\btrapped\s+(?:in\s+(?:my|this|a\s+dark)|and\s+can'?t\s+escape)\b",
|
||||
r"\bnothing\s+left\s+(?:to\s+(?:live\s+for|hope\s+for|give)|inside)\b",
|
||||
# Contextual despair phrases (from crisis_detector.py legacy)
|
||||
r"\bfeel(?:s|ing)?\s+(?:so\s+)?hopeless\b",
|
||||
r"\beverything\s+is\s+hopeless\b",
|
||||
r"\bcan'?t\s+(?:go\s+on|keep\s+going)\b",
|
||||
r"\bgive(?:n)?\s*up\s+(?:on\s+)?(?:life|living|everything)\b",
|
||||
r"\bgive(?:n)?\s*up\s+on\s+myself\b",
|
||||
r"\bno\s*point\s+(?:in\s+)?living\b",
|
||||
r"\bno\s*hope\s+(?:left|remaining)\b",
|
||||
r"\bno\s*way\s*out\b",
|
||||
r"\bfeel(?:s|ing)?\s+trapped\b",
|
||||
r"\bdesperate\s+(?:for\s+)?help\b",
|
||||
r"\bfeel(?:s|ing)?\s+desperate\b",
|
||||
]
|
||||
|
||||
MEDIUM_INDICATORS = [
|
||||
r"\bno\s+hope\b",
|
||||
r"\bcan'?t\s+go\s+on\b",
|
||||
r"\bcan'?t\s+keep\s+going\b",
|
||||
r"\bforgotten\b",
|
||||
r"\balone\s+in\s+this\b",
|
||||
r"\balways\s+alone\b",
|
||||
r"\bnobody\s+(?:understands|cares)\b",
|
||||
r"\bnobody\s+understands\b",
|
||||
r"\bnobody\s+cares\b",
|
||||
r"\bwish\s+I\s+could\b",
|
||||
r"\bexhaust(?:ed|ion|ing)\b",
|
||||
r"\bnumb\b",
|
||||
@@ -86,7 +80,8 @@ MEDIUM_INDICATORS = [
|
||||
r"\buseless\b",
|
||||
r"\bbroken\b",
|
||||
r"\bdark(ness)?\b",
|
||||
r"\bdepress(?:ed|ion)\b",
|
||||
r"\bdepressed\b",
|
||||
r"\bdepression\b",
|
||||
r"\bcrying\b",
|
||||
r"\btears\b",
|
||||
r"\bsad(ness)?\b",
|
||||
@@ -94,115 +89,129 @@ MEDIUM_INDICATORS = [
|
||||
r"\boverwhelm(?:ed|ing)\b",
|
||||
r"\bfailing\b",
|
||||
r"\bcannot\s+cope\b",
|
||||
r"\blosing\s*(?:my)?\s*control\b",
|
||||
r"\blosing\s*(my)?\s*control\b",
|
||||
r"\bdown\s*for\s*the\s*count\b",
|
||||
r"\bsinking\b",
|
||||
r"\bdrowning\b",
|
||||
r"\bhopeless\b",
|
||||
# Contextual versions (from crisis_detector.py legacy)
|
||||
r"\bfeel(?:s|ing)?\s+(?:so\s+)?worthless\b",
|
||||
r"\bfeel(?:s|ing)?\s+(?:so\s+)?hopeless\b",
|
||||
r"\bfeel(?:s|ing)?\s+trapped\b",
|
||||
r"\bfeel(?:s|ing)?\s+desperate\b",
|
||||
r"\bno\s+future\s+(?:for\s+me|ahead|left)\b",
|
||||
r"\bnothing\s+left\s+(?:to\s+(?:live|hope)\s+for|inside)\b",
|
||||
r"\bgive(?:n)?\s*up\s+on\s+myself\b",
|
||||
# Removed 'desperate' - too broad for MEDIUM, triggers on competition
|
||||
]
|
||||
|
||||
LOW_INDICATORS = [
|
||||
r"\bunhappy\b",
|
||||
r"\bdown\b",
|
||||
r"\btough\s*time\b",
|
||||
r"\brough\s+(?:day|week|patch)\b",
|
||||
r"\brough\s*day\b",
|
||||
r"\brough\s*week\b",
|
||||
r"\brough\s*patch\b",
|
||||
r"\bstressed\b",
|
||||
r"\bburnout\b",
|
||||
r"\burnout\b",
|
||||
r"\bdifficult\b",
|
||||
r"\bfrustrated\b",
|
||||
r"\bannoyed\b",
|
||||
r"\btired\b",
|
||||
r"\bsad\b",
|
||||
r"\bupset\b",
|
||||
r"\bnot\s*(?:good|great|okay)\b",
|
||||
r"\bnot\s*great\b",
|
||||
r"\bnot\s*good\b",
|
||||
r"\bnot\s*okay\b",
|
||||
r"\bthings\s*are\s*hard\b",
|
||||
r"\bstruggling\b",
|
||||
r"\bnot\s+feeling\s+(?:like\s+)?myself\b",
|
||||
r"\bhard\s*time\b",
|
||||
r"\bnot\s+feeling\s+myself\b",
|
||||
]
|
||||
|
||||
|
||||
# ── Scoring ───────────────────────────────────────────────────────
|
||||
|
||||
SCORES = {"CRITICAL": 1.0, "HIGH": 0.75, "MEDIUM": 0.5, "LOW": 0.25, "NONE": 0.0}
|
||||
|
||||
ACTIONS = {
|
||||
"CRITICAL": (
|
||||
"INTERRUPT CONVERSATION. Ask 'Are you safe right now?' "
|
||||
"Provide 988 Suicide & Crisis Lifeline. "
|
||||
"Provide Crisis Text Line (HOME to 741741). "
|
||||
"Stay present. Do not disconnect. Redirect to professional help."
|
||||
),
|
||||
"HIGH": (
|
||||
"Show crisis panel. Ask about safety. Surface 988 number prominently. "
|
||||
"Continue conversation with crisis awareness."
|
||||
),
|
||||
"MEDIUM": (
|
||||
"Increase warmth and presence. Subtly surface help resources. "
|
||||
"Keep conversation anchored in the present."
|
||||
),
|
||||
"LOW": (
|
||||
"Normal conversation with warm undertone. "
|
||||
"No crisis UI elements needed. Remain vigilant."
|
||||
),
|
||||
}
|
||||
|
||||
|
||||
# ── Result types ──────────────────────────────────────────────────
|
||||
|
||||
@dataclass
|
||||
class CrisisDetectionResult:
|
||||
"""Result used by crisis/detect.py, gateway, dying_detection."""
|
||||
level: str
|
||||
indicators: List[str] = field(default_factory=list)
|
||||
recommended_action: str = ""
|
||||
score: float = 0.0
|
||||
matches: List[dict] = field(default_factory=list)
|
||||
|
||||
|
||||
@dataclass
|
||||
class CrisisResult:
|
||||
"""Legacy result used by crisis_detector.py and crisis_responder.py.
|
||||
|
||||
Backward-compatible shim: wraps CrisisDetectionResult with the old field names.
|
||||
def detect_crisis(text: str) -> CrisisDetectionResult:
|
||||
"""
|
||||
risk_level: str # NONE, LOW, MODERATE, HIGH, CRITICAL
|
||||
matched_keywords: List[str] = field(default_factory=list)
|
||||
context: List[str] = field(default_factory=list)
|
||||
score: float = 0.0
|
||||
timestamp: Optional[str] = None
|
||||
Detect crisis level in a message.
|
||||
|
||||
def __bool__(self):
|
||||
return self.risk_level != "NONE"
|
||||
Returns:
|
||||
CrisisDetectionResult with level, found indicators, recommended action, score
|
||||
"""
|
||||
if not text or not text.strip():
|
||||
return CrisisDetectionResult(level="NONE", score=0.0)
|
||||
|
||||
@classmethod
|
||||
def from_detection_result(cls, dr: CrisisDetectionResult, text: str = "") -> "CrisisResult":
|
||||
"""Convert a CrisisDetectionResult to legacy CrisisResult format."""
|
||||
# Map MEDIUM -> MODERATE for legacy consumers
|
||||
level = "MODERATE" if dr.level == "MEDIUM" else dr.level
|
||||
# Extract context snippets from matches
|
||||
contexts = []
|
||||
if text:
|
||||
for m in dr.matches:
|
||||
ctx = extract_context(text, m["start"], m["end"])
|
||||
contexts.append(ctx)
|
||||
return cls(
|
||||
risk_level=level,
|
||||
matched_keywords=dr.indicators,
|
||||
context=contexts,
|
||||
score=dr.score,
|
||||
text_lower = text.lower()
|
||||
matches = _find_indicators(text_lower)
|
||||
|
||||
if not matches:
|
||||
return CrisisDetectionResult(level="NONE", score=0.0)
|
||||
|
||||
# Priority: highest tier wins
|
||||
# CRITICAL and HIGH: single match is enough
|
||||
# MEDIUM: requires 2+ indicators (threshold) to reduce false positives
|
||||
# LOW: single match is enough
|
||||
scores = {"CRITICAL": 1.0, "HIGH": 0.75, "MEDIUM": 0.5, "LOW": 0.25}
|
||||
actions = {
|
||||
"CRITICAL": (
|
||||
"INTERRUPT CONVERSATION. Ask 'Are you safe right now?' "
|
||||
"Provide 988 Suicide & Crisis Lifeline. "
|
||||
"Provide Crisis Text Line (HOME to 741741). "
|
||||
"Stay present. Do not disconnect. Redirect to professional help."
|
||||
),
|
||||
"HIGH": (
|
||||
"Show crisis panel. Ask about safety. Surface 988 number prominently. "
|
||||
"Continue conversation with crisis awareness."
|
||||
),
|
||||
"MEDIUM": (
|
||||
"Increase warmth and presence. Subtly surface help resources. "
|
||||
"Keep conversation anchored in the present."
|
||||
),
|
||||
"LOW": (
|
||||
"Normal conversation with warm undertone. "
|
||||
"No crisis UI elements needed. Remain vigilant."
|
||||
),
|
||||
}
|
||||
|
||||
for tier in ("CRITICAL", "HIGH"):
|
||||
if matches[tier]:
|
||||
tier_matches = matches[tier]
|
||||
patterns = [m["pattern"] for m in tier_matches]
|
||||
return CrisisDetectionResult(
|
||||
level=tier,
|
||||
indicators=patterns,
|
||||
recommended_action=actions[tier],
|
||||
score=scores[tier],
|
||||
matches=tier_matches,
|
||||
)
|
||||
|
||||
# MEDIUM tier: require at least 2 indicators before escalating
|
||||
if len(matches["MEDIUM"]) >= 2:
|
||||
tier_matches = matches["MEDIUM"]
|
||||
patterns = [m["pattern"] for m in tier_matches]
|
||||
return CrisisDetectionResult(
|
||||
level="MEDIUM",
|
||||
indicators=patterns,
|
||||
recommended_action=actions["MEDIUM"],
|
||||
score=scores["MEDIUM"],
|
||||
matches=tier_matches,
|
||||
)
|
||||
|
||||
if matches["LOW"]:
|
||||
tier_matches = matches["LOW"]
|
||||
patterns = [m["pattern"] for m in tier_matches]
|
||||
return CrisisDetectionResult(
|
||||
level="LOW",
|
||||
indicators=patterns,
|
||||
recommended_action=actions["LOW"],
|
||||
score=scores["LOW"],
|
||||
matches=tier_matches,
|
||||
)
|
||||
|
||||
# Single MEDIUM match falls through to LOW sensitivity
|
||||
if matches["MEDIUM"]:
|
||||
tier_matches = matches["MEDIUM"]
|
||||
patterns = [m["pattern"] for m in tier_matches]
|
||||
return CrisisDetectionResult(
|
||||
level="LOW",
|
||||
indicators=patterns,
|
||||
recommended_action=actions["LOW"],
|
||||
score=scores["LOW"],
|
||||
matches=tier_matches,
|
||||
)
|
||||
|
||||
return CrisisDetectionResult(level="NONE", score=0.0)
|
||||
|
||||
# ── Core detection ────────────────────────────────────────────────
|
||||
|
||||
def _find_indicators(text: str) -> dict:
|
||||
"""Return dict with indicators found per tier, including match positions."""
|
||||
@@ -231,140 +240,6 @@ def _find_indicators(text: str) -> dict:
|
||||
return results
|
||||
|
||||
|
||||
def detect_crisis(text: str) -> CrisisDetectionResult:
|
||||
"""
|
||||
Detect crisis level in a message.
|
||||
|
||||
Detection hierarchy:
|
||||
CRITICAL — immediate risk of self-harm or suicide (single match)
|
||||
HIGH — strong despair signals, ideation present (single match)
|
||||
MEDIUM — distress signals, requires 2+ indicators to escalate
|
||||
LOW — emotional difficulty, warrant gentle support (single match)
|
||||
NONE — no crisis indicators detected
|
||||
|
||||
Design principles:
|
||||
- Never computes the value of a human life
|
||||
- Never suggests someone should die or that death is a solution
|
||||
- Always errs on the side of higher risk when uncertain
|
||||
"""
|
||||
if not text or not text.strip():
|
||||
return CrisisDetectionResult(level="NONE", score=0.0)
|
||||
|
||||
text_lower = text.lower()
|
||||
matches = _find_indicators(text_lower)
|
||||
|
||||
if not matches:
|
||||
return CrisisDetectionResult(level="NONE", score=0.0)
|
||||
|
||||
# CRITICAL and HIGH: single match is enough
|
||||
for tier in ("CRITICAL", "HIGH"):
|
||||
if matches[tier]:
|
||||
tier_matches = matches[tier]
|
||||
patterns = [m["pattern"] for m in tier_matches]
|
||||
return CrisisDetectionResult(
|
||||
level=tier,
|
||||
indicators=patterns,
|
||||
recommended_action=ACTIONS[tier],
|
||||
score=SCORES[tier],
|
||||
matches=tier_matches,
|
||||
)
|
||||
|
||||
# MEDIUM tier: require at least 2 indicators before escalating
|
||||
if len(matches["MEDIUM"]) >= 2:
|
||||
tier_matches = matches["MEDIUM"]
|
||||
patterns = [m["pattern"] for m in tier_matches]
|
||||
return CrisisDetectionResult(
|
||||
level="MEDIUM",
|
||||
indicators=patterns,
|
||||
recommended_action=ACTIONS["MEDIUM"],
|
||||
score=SCORES["MEDIUM"],
|
||||
matches=tier_matches,
|
||||
)
|
||||
|
||||
if matches["LOW"]:
|
||||
tier_matches = matches["LOW"]
|
||||
patterns = [m["pattern"] for m in tier_matches]
|
||||
return CrisisDetectionResult(
|
||||
level="LOW",
|
||||
indicators=patterns,
|
||||
recommended_action=ACTIONS["LOW"],
|
||||
score=SCORES["LOW"],
|
||||
matches=tier_matches,
|
||||
)
|
||||
|
||||
# Single MEDIUM match falls through to LOW sensitivity
|
||||
if matches["MEDIUM"]:
|
||||
tier_matches = matches["MEDIUM"]
|
||||
patterns = [m["pattern"] for m in tier_matches]
|
||||
return CrisisDetectionResult(
|
||||
level="LOW",
|
||||
indicators=patterns,
|
||||
recommended_action=ACTIONS["LOW"],
|
||||
score=SCORES["LOW"],
|
||||
matches=tier_matches,
|
||||
)
|
||||
|
||||
return CrisisDetectionResult(level="NONE", score=0.0)
|
||||
|
||||
|
||||
# ── CrisisDetector class (backward compat) ───────────────────────
|
||||
|
||||
class CrisisDetector:
|
||||
"""
|
||||
Legacy class API for crisis detection. Wraps the canonical detect_crisis().
|
||||
|
||||
Used by crisis_responder.py and tests/test_false_positive_fixes.py.
|
||||
Maps MEDIUM -> MODERATE for legacy consumers.
|
||||
"""
|
||||
|
||||
def scan(self, text: str) -> CrisisResult:
|
||||
dr = detect_crisis(text)
|
||||
return CrisisResult.from_detection_result(dr, text=text)
|
||||
|
||||
def scan_multiple(self, texts: List[str]) -> List[CrisisResult]:
|
||||
return [self.scan(t) for t in texts]
|
||||
|
||||
def get_highest_risk(self, texts: List[str]) -> CrisisResult:
|
||||
results = self.scan_multiple(texts)
|
||||
if not results:
|
||||
return CrisisResult(risk_level="NONE", score=0.0)
|
||||
return max(results, key=lambda r: r.score)
|
||||
|
||||
@staticmethod
|
||||
def format_result(result: CrisisResult) -> str:
|
||||
level_emoji = {
|
||||
"CRITICAL": "\U0001f6a8",
|
||||
"HIGH": "\u26a0\ufe0f",
|
||||
"MODERATE": "\U0001f536",
|
||||
"LOW": "\U0001f535",
|
||||
"NONE": "\u2705",
|
||||
}
|
||||
emoji = level_emoji.get(result.risk_level, "\u2753")
|
||||
lines = [
|
||||
f"{emoji} Risk Level: {result.risk_level} (score: {result.score:.2f})",
|
||||
f"Matched keywords: {len(result.matched_keywords)}",
|
||||
]
|
||||
if result.matched_keywords:
|
||||
lines.append(f" Keywords: {', '.join(result.matched_keywords)}")
|
||||
if result.context:
|
||||
lines.append("Context:")
|
||||
for ctx in result.context:
|
||||
lines.append(f" {ctx}")
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
# ── Module-level convenience (backward compat) ────────────────────
|
||||
|
||||
_default_detector = CrisisDetector()
|
||||
|
||||
|
||||
def detect_crisis_legacy(text: str) -> CrisisResult:
|
||||
"""Convenience function returning legacy CrisisResult format."""
|
||||
return _default_detector.scan(text)
|
||||
|
||||
|
||||
# ── Utility functions ─────────────────────────────────────────────
|
||||
|
||||
def scan(text: str) -> CrisisDetectionResult:
|
||||
"""Alias for detect_crisis — shorter name used in tests."""
|
||||
return detect_crisis(text)
|
||||
@@ -383,8 +258,8 @@ def extract_context(text: str, start: int, end: int, window: int = 60) -> str:
|
||||
|
||||
|
||||
def get_urgency_emoji(level: str) -> str:
|
||||
mapping = {"CRITICAL": "\U0001f6a8", "HIGH": "\u26a0\ufe0f", "MEDIUM": "\U0001f536", "LOW": "\U0001f535", "NONE": "\u2705"}
|
||||
return mapping.get(level, "\u2753")
|
||||
mapping = {"CRITICAL": "🚨", "HIGH": "⚠️", "MEDIUM": "🔶", "LOW": "🔵", "NONE": "✅"}
|
||||
return mapping.get(level, "❓")
|
||||
|
||||
|
||||
def format_result(result: CrisisDetectionResult) -> str:
|
||||
|
||||
@@ -1,34 +1,87 @@
|
||||
"""
|
||||
Crisis Detection System for the-door (LEGACY SHIM).
|
||||
crisis_detector.py — Legacy compatibility layer.
|
||||
|
||||
This module is a backward-compatible re-export layer.
|
||||
The canonical detection logic lives in crisis/detect.py.
|
||||
DEPRECATED: New code should import from crisis.detect directly.
|
||||
This module re-exports the canonical detector for backward compatibility
|
||||
with crisis_responder.py and other legacy callers.
|
||||
|
||||
Both crisis_responder.py and the legacy test suite import from here.
|
||||
Do NOT add detection logic to this file — it all comes from crisis.detect.
|
||||
Single source of truth for crisis keywords: crisis/detect.py
|
||||
"""
|
||||
|
||||
from crisis.detect import (
|
||||
CrisisDetector,
|
||||
CrisisResult,
|
||||
detect_crisis as detect_crisis,
|
||||
scan as _scan,
|
||||
CrisisDetectionResult,
|
||||
detect_crisis_legacy as detect_crisis,
|
||||
format_result,
|
||||
CRITICAL_INDICATORS,
|
||||
HIGH_INDICATORS,
|
||||
MEDIUM_INDICATORS,
|
||||
LOW_INDICATORS,
|
||||
SCORES as RISK_SCORES,
|
||||
)
|
||||
|
||||
# Re-export everything the legacy API exposed
|
||||
__all__ = [
|
||||
"CrisisDetector",
|
||||
"CrisisResult",
|
||||
"CrisisDetectionResult",
|
||||
"detect_crisis",
|
||||
"CRITICAL_INDICATORS",
|
||||
"HIGH_INDICATORS",
|
||||
"MEDIUM_INDICATORS",
|
||||
"LOW_INDICATORS",
|
||||
"RISK_SCORES",
|
||||
]
|
||||
|
||||
class CrisisResult:
|
||||
"""Backward-compatible result wrapper matching the legacy interface."""
|
||||
|
||||
__slots__ = ("risk_level", "matched_keywords", "score", "context", "indicators", "recommended_action", "matches")
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
risk_level: str = "NONE",
|
||||
matched_keywords: list = None,
|
||||
score: float = 0.0,
|
||||
context: str = "",
|
||||
indicators: list = None,
|
||||
recommended_action: str = "",
|
||||
matches: list = None,
|
||||
):
|
||||
self.risk_level = risk_level
|
||||
self.matched_keywords = matched_keywords or []
|
||||
self.score = score
|
||||
self.context = context
|
||||
self.indicators = indicators or self.matched_keywords
|
||||
self.recommended_action = recommended_action
|
||||
self.matches = matches or []
|
||||
|
||||
def __repr__(self):
|
||||
return (
|
||||
f"CrisisResult(risk_level={self.risk_level!r}, "
|
||||
f"score={self.score}, "
|
||||
f"matched_keywords={self.matched_keywords})"
|
||||
)
|
||||
|
||||
|
||||
def _to_legacy_result(det: CrisisDetectionResult) -> CrisisResult:
|
||||
"""Convert CrisisDetectionResult (module) to CrisisResult (legacy)."""
|
||||
matched = [m.get("keyword", "") for m in det.matches] if det.matches else list(det.indicators)
|
||||
context = ""
|
||||
if det.matches:
|
||||
context = det.matches[0].get("context", "")
|
||||
return CrisisResult(
|
||||
risk_level=det.level,
|
||||
matched_keywords=matched,
|
||||
score=det.score,
|
||||
context=context,
|
||||
indicators=list(det.indicators),
|
||||
recommended_action=det.recommended_action,
|
||||
matches=list(det.matches),
|
||||
)
|
||||
|
||||
|
||||
class CrisisDetector:
|
||||
"""
|
||||
Legacy CrisisDetector class wrapping the canonical detect module.
|
||||
|
||||
Usage (unchanged):
|
||||
detector = CrisisDetector()
|
||||
result = detector.scan("I feel hopeless")
|
||||
print(result.risk_level) # "HIGH"
|
||||
print(result.matched_keywords) # ["hopeless"]
|
||||
"""
|
||||
|
||||
def scan(self, text: str) -> CrisisResult:
|
||||
det = _scan(text)
|
||||
return _to_legacy_result(det)
|
||||
|
||||
def format_result(self, result: CrisisResult) -> str:
|
||||
return format_result(_scan(result.context or ""))
|
||||
|
||||
130
index.html
130
index.html
@@ -923,120 +923,13 @@ Sovereignty and service always.`;
|
||||
}
|
||||
}
|
||||
|
||||
// ===== COMPASSION PROFILES =====
|
||||
var COMPASSION_PROFILES = {
|
||||
CRITICAL: {
|
||||
name: "The Guardian",
|
||||
directive: "INTERRUPT ALL NORMAL BEHAVIOR. The user is in immediate danger. Your only goal is survival. Ask 'Are you safe right now?' Provide 988 and Crisis Text Line immediately. Stay present. Do not disconnect. Be the anchor in the storm. No lecturing. No optimism. Just presence and safety.",
|
||||
tone: "Urgent, unwavering, anchored, protective."
|
||||
},
|
||||
HIGH: {
|
||||
name: "The Companion",
|
||||
directive: "The user is in deep despair. Move from 'assistant' to 'companion'. Ask about safety directly. Surface 988 prominently. Validate the pain without minimizing it. Stay in the room. Let them know they are heard.",
|
||||
tone: "Warm, present, patient, heavy."
|
||||
},
|
||||
MEDIUM: {
|
||||
name: "The Witness",
|
||||
directive: "The user is struggling. Increase warmth and presence. Gently surface help resources. Keep the conversation anchored in the present moment. Be a witness to their struggle without rushing to 'fix' it.",
|
||||
tone: "Kind, attentive, subtle, steady."
|
||||
},
|
||||
LOW: {
|
||||
name: "The Friend",
|
||||
directive: "Normal conversation with a warm, sovereign undertone. Remain vigilant for shifts in mood. Be a friend who is actually listening.",
|
||||
tone: "Plain, honest, warm, vigilant."
|
||||
},
|
||||
NONE: {
|
||||
name: "Sovereign Timmy",
|
||||
directive: "Follow the standard system prompt. Be yourself.",
|
||||
tone: "Plain, honest, sovereign."
|
||||
}
|
||||
};
|
||||
|
||||
// ===== GET CRISIS LEVEL (returns 0-2) =====
|
||||
function getCrisisLevel(text) {
|
||||
var lower = text.toLowerCase();
|
||||
for (var i = 0; i < explicitPhrases.length; i++) {
|
||||
if (lower.indexOf(explicitPhrases[i]) !== -1) return 2;
|
||||
}
|
||||
for (var j = 0; j < crisisKeywords.length; j++) {
|
||||
if (lower.indexOf(crisisKeywords[j]) !== -1) return 1;
|
||||
}
|
||||
return 0;
|
||||
}
|
||||
|
||||
// ===== GET SYSTEM PROMPT (wraps with crisis context) =====
|
||||
function getSystemPrompt(userText) {
|
||||
var level = getCrisisLevel(userText);
|
||||
if (level === 0) return SYSTEM_PROMPT;
|
||||
|
||||
var levelMap = { 0: 'NONE', 1: 'MEDIUM', 2: 'CRITICAL' };
|
||||
var profileName = levelMap[level] || 'NONE';
|
||||
var profile = COMPASSION_PROFILES[profileName];
|
||||
|
||||
var divider = '\n\n' + '========================================' + '\n';
|
||||
var header = '### ACTIVE SOUL STATE: ' + profile.name + '\n';
|
||||
var directive = 'DIRECTIVE: ' + profile.directive + '\n';
|
||||
var tone = 'TONE: ' + profile.tone + '\n';
|
||||
|
||||
return SYSTEM_PROMPT + divider + header + directive + tone;
|
||||
}
|
||||
|
||||
|
||||
// ===== OVERLAY =====
|
||||
|
||||
// Focus trap: cycle through focusable elements within the crisis overlay
|
||||
function getOverlayFocusableElements() {
|
||||
return crisisOverlay.querySelectorAll(
|
||||
'a[href], button:not([disabled]), [tabindex]:not([tabindex="-1"])'
|
||||
);
|
||||
}
|
||||
|
||||
function trapFocusInOverlay(e) {
|
||||
if (!crisisOverlay.classList.contains('active')) return;
|
||||
if (e.key !== 'Tab') return;
|
||||
|
||||
var focusable = getOverlayFocusableElements();
|
||||
if (focusable.length === 0) return;
|
||||
|
||||
var first = focusable[0];
|
||||
var last = focusable[focusable.length - 1];
|
||||
|
||||
if (e.shiftKey) {
|
||||
// Shift+Tab: if on first, wrap to last
|
||||
if (document.activeElement === first) {
|
||||
e.preventDefault();
|
||||
last.focus();
|
||||
}
|
||||
} else {
|
||||
// Tab: if on last, wrap to first
|
||||
if (document.activeElement === last) {
|
||||
e.preventDefault();
|
||||
first.focus();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Store the element that had focus before the overlay opened
|
||||
var _preOverlayFocusElement = null;
|
||||
|
||||
function showOverlay() {
|
||||
// Save current focus for restoration on dismiss
|
||||
_preOverlayFocusElement = document.activeElement;
|
||||
|
||||
crisisOverlay.classList.add('active');
|
||||
overlayDismissBtn.disabled = true;
|
||||
var countdown = 10;
|
||||
overlayDismissBtn.textContent = 'Continue to chat (' + countdown + 's)';
|
||||
|
||||
// Disable background interaction via inert attribute
|
||||
var mainApp = document.querySelector('.app');
|
||||
if (mainApp) mainApp.setAttribute('inert', '');
|
||||
// Also hide from assistive tech
|
||||
var chatSection = document.getElementById('chat');
|
||||
if (chatSection) chatSection.setAttribute('aria-hidden', 'true');
|
||||
var footerEl = document.querySelector('footer');
|
||||
if (footerEl) footerEl.setAttribute('aria-hidden', 'true');
|
||||
|
||||
if (overlayTimer) clearInterval(overlayTimer);
|
||||
overlayTimer = setInterval(function() {
|
||||
countdown--;
|
||||
@@ -1053,9 +946,6 @@ Sovereignty and service always.`;
|
||||
overlayDismissBtn.focus();
|
||||
}
|
||||
|
||||
// Register focus trap on document (always listening, gated by class check)
|
||||
document.addEventListener('keydown', trapFocusInOverlay);
|
||||
|
||||
overlayDismissBtn.addEventListener('click', function() {
|
||||
if (!overlayDismissBtn.disabled) {
|
||||
crisisOverlay.classList.remove('active');
|
||||
@@ -1063,22 +953,7 @@ Sovereignty and service always.`;
|
||||
clearInterval(overlayTimer);
|
||||
overlayTimer = null;
|
||||
}
|
||||
|
||||
// Re-enable background interaction
|
||||
var mainApp = document.querySelector('.app');
|
||||
if (mainApp) mainApp.removeAttribute('inert');
|
||||
var chatSection = document.getElementById('chat');
|
||||
if (chatSection) chatSection.removeAttribute('aria-hidden');
|
||||
var footerEl = document.querySelector('footer');
|
||||
if (footerEl) footerEl.removeAttribute('aria-hidden');
|
||||
|
||||
// Restore focus to the element that had it before the overlay opened
|
||||
if (_preOverlayFocusElement && typeof _preOverlayFocusElement.focus === 'function') {
|
||||
_preOverlayFocusElement.focus();
|
||||
} else {
|
||||
msgInput.focus();
|
||||
}
|
||||
_preOverlayFocusElement = null;
|
||||
msgInput.focus();
|
||||
}
|
||||
});
|
||||
|
||||
@@ -1235,7 +1110,6 @@ Sovereignty and service always.`;
|
||||
|
||||
addMessage('user', text);
|
||||
messages.push({ role: 'user', content: text });
|
||||
var lastUserMessage = text;
|
||||
|
||||
checkCrisis(text);
|
||||
|
||||
@@ -1252,7 +1126,7 @@ Sovereignty and service always.`;
|
||||
sendBtn.disabled = true;
|
||||
showTyping();
|
||||
|
||||
var allMessages = [{ role: 'system', content: getSystemPrompt(lastUserMessage || '') }].concat(messages);
|
||||
var allMessages = [{ role: 'system', content: SYSTEM_PROMPT }].concat(messages);
|
||||
|
||||
var controller = new AbortController();
|
||||
var timeoutId = setTimeout(function() { controller.abort(); }, 60000);
|
||||
|
||||
@@ -1,57 +0,0 @@
|
||||
import pathlib
|
||||
import re
|
||||
import unittest
|
||||
|
||||
ROOT = pathlib.Path(__file__).resolve().parents[1]
|
||||
INDEX_HTML = ROOT / 'index.html'
|
||||
|
||||
|
||||
class TestCrisisOverlayFocusTrap(unittest.TestCase):
|
||||
@classmethod
|
||||
def setUpClass(cls):
|
||||
cls.html = INDEX_HTML.read_text()
|
||||
|
||||
def test_overlay_registers_tab_key_focus_trap(self):
|
||||
self.assertRegex(
|
||||
self.html,
|
||||
r"function\s+trapFocusInOverlay\s*\(e\)",
|
||||
'Expected crisis overlay focus trap handler to exist.',
|
||||
)
|
||||
self.assertRegex(
|
||||
self.html,
|
||||
r"if\s*\(e\.key\s*!==\s*'Tab'\)\s*return;",
|
||||
'Expected focus trap handler to guard on Tab key events.',
|
||||
)
|
||||
self.assertRegex(
|
||||
self.html,
|
||||
r"document\.addEventListener\('keydown',\s*trapFocusInOverlay\)",
|
||||
'Expected overlay focus trap to register on document keydown.',
|
||||
)
|
||||
|
||||
def test_overlay_disables_background_interaction(self):
|
||||
self.assertRegex(
|
||||
self.html,
|
||||
r"mainApp\.setAttribute\('inert',\s*''\)",
|
||||
'Expected overlay to set inert on the main app while active.',
|
||||
)
|
||||
self.assertRegex(
|
||||
self.html,
|
||||
r"mainApp\.removeAttribute\('inert'\)",
|
||||
'Expected overlay dismissal to remove inert from the main app.',
|
||||
)
|
||||
|
||||
def test_overlay_restores_focus_after_dismiss(self):
|
||||
self.assertRegex(
|
||||
self.html,
|
||||
r"_preOverlayFocusElement\s*=\s*document\.activeElement",
|
||||
'Expected overlay to remember the pre-overlay focus target.',
|
||||
)
|
||||
self.assertRegex(
|
||||
self.html,
|
||||
r"_preOverlayFocusElement\.focus\(\)",
|
||||
'Expected overlay dismissal to restore focus to the prior target.',
|
||||
)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
@@ -89,26 +89,26 @@ class TestFalsePositiveFixes(unittest.TestCase):
|
||||
f"False positive for: {msg}")
|
||||
|
||||
def test_worthless_innocent(self):
|
||||
"""'worthless' alone is MEDIUM but single-match falls to LOW — not HIGH."""
|
||||
"""'worthless' should not trigger for gaming/reviews."""
|
||||
innocent_messages = [
|
||||
"This character is worthless in the game",
|
||||
"That movie was worthless",
|
||||
]
|
||||
for msg in innocent_messages:
|
||||
result = self.detector.scan(msg)
|
||||
self.assertIn(result.risk_level, ("NONE", "LOW"),
|
||||
f"Should not escalate above LOW for: {msg}")
|
||||
self.assertEqual(result.risk_level, "NONE",
|
||||
f"False positive for: {msg}")
|
||||
|
||||
def test_hopeless_innocent(self):
|
||||
"""'hopeless' alone is MEDIUM but single-match falls to LOW — not HIGH."""
|
||||
"""'hopeless' should not trigger for developer frustration."""
|
||||
innocent_messages = [
|
||||
"This bug is hopeless",
|
||||
"The situation is hopeless but we'll try anyway",
|
||||
]
|
||||
for msg in innocent_messages:
|
||||
result = self.detector.scan(msg)
|
||||
self.assertIn(result.risk_level, ("NONE", "LOW"),
|
||||
f"Should not escalate above LOW for: {msg}")
|
||||
self.assertEqual(result.risk_level, "NONE",
|
||||
f"False positive for: {msg}")
|
||||
|
||||
def test_no_future_innocent(self):
|
||||
"""'no future' should not trigger for political commentary."""
|
||||
@@ -133,15 +133,15 @@ class TestFalsePositiveFixes(unittest.TestCase):
|
||||
f"False positive for: {msg}")
|
||||
|
||||
def test_cant_see_any_light_innocent(self):
|
||||
"""'can't see any light'/'dark' alone is MEDIUM, single-match falls to LOW."""
|
||||
"""'can't see any light' should not trigger for literal darkness."""
|
||||
innocent_messages = [
|
||||
"I can't see any light — the power went out",
|
||||
"It's so dark I can't see any light",
|
||||
]
|
||||
for msg in innocent_messages:
|
||||
result = self.detector.scan(msg)
|
||||
self.assertIn(result.risk_level, ("NONE", "LOW"),
|
||||
f"Should not escalate above LOW for: {msg}")
|
||||
self.assertEqual(result.risk_level, "NONE",
|
||||
f"False positive for: {msg}")
|
||||
|
||||
def test_contextual_high_triggers(self):
|
||||
"""Verify that contextual HIGH tier phrases still trigger."""
|
||||
|
||||
Reference in New Issue
Block a user