Compare commits

...

12 Commits

Author SHA1 Message Date
Alexander Whitestone
cb2c325050 feat: session-level crisis tracking and escalation (closes #35)
Some checks failed
Sanity Checks / sanity-test (pull_request) Successful in 3s
Smoke Test / smoke (pull_request) Failing after 1m37s
Implement CrisisSessionTracker for per-session crisis state management.

Tracks crisis levels across conversation messages to detect:
- Escalation: LOW -> HIGH in N messages triggers heightened awareness
- De-escalation: CRITICAL -> LOW sustained for 5+ messages confirms stepping down
- Sustained elevated levels: HIGH/CRITICAL maintained over multiple messages

State tracked per session:
- current_level, peak_level, message_count, level_history
- escalated flag, deescalation_count, deescalation_confirmed

Privacy-first: in-memory only, resets on new session, never persists to disk.

System prompt context injection: generates human-readable context strings
describing the session's crisis trajectory for AI awareness.

120 tests (36 new + 84 existing) all pass.
2026-04-13 17:28:41 -04:00
Alexander Whitestone
b7e10c7f15 feat: Session-level crisis tracking and escalation (#35)
All checks were successful
Sanity Checks / sanity-test (pull_request) Successful in 3s
Smoke Test / smoke (pull_request) Successful in 4s
Tracks crisis level across messages within a conversation.
Detects escalation (getting worse) and de-escalation (getting better).
Adds session context to system prompt so AI sees the trajectory.

State:
- currentLevel: max level seen this session
- peakLevel: highest level ever reached
- messageCount: total messages
- history: [{level, timestamp}] per message
- escalationRate: levels per message over last 5

Detection:
- Escalation: level rising >0.5/msg over 3 messages → alert
- De-escalation: peak CRITICAL, current LOW, 5+ messages → step down
- Persistent: HIGH+ for 3+ messages → sustained crisis note
- Trajectory: level history shown as 0 → 1 → 2 → 0

Privacy:
- Session-only state, no persistence across sessions
- No database, no logging, no cross-conversation tracking
- resetSessionCrisis() on new session

Tests: 84 passed

Fixes #35
2026-04-13 16:04:55 -04:00
9b94978d1c feat: Wire compassion router into gateway flow (#34) (#43)
All checks were successful
Smoke Test / smoke (push) Successful in 5s
Squash merge: wire compassion router into gateway flow
2026-04-13 19:59:15 +00:00
e71bca1744 fix: de-duplicate crisis_detector.py and crisis/detect.py (closes #39) (#44)
Some checks failed
Smoke Test / smoke (push) Has been cancelled
Squash merge: de-duplicate crisis detector (closes #39)

Co-authored-by: Alexander Whitestone <alexander@alexanderwhitestone.com>
Co-committed-by: Alexander Whitestone <alexander@alexanderwhitestone.com>
2026-04-13 19:59:12 +00:00
Alexander Whitestone
1d8afc30fd fix: reduce crisis detector false positives (closes #32)
All checks were successful
Smoke Test / smoke (push) Successful in 5s
- Removed 'saying goodbye' from CRITICAL tier (too common in innocent contexts)
- Narrowed 'wrote a will' to 'wrote a suicide note' (responsible behavior)
- Removed broad single-word matches from HIGH tier: give up, trapped, desperate, worthless, hopeless, no future, nothing left, can't see any light
- Added contextual HIGH tier phrases: feel hopeless, trapped in this, desperate for help, give up on life, etc.
- Updated MODERATE tier with contextual versions: feel worthless, feel hopeless, feel trapped, etc.
- Updated index.html JavaScript keywords to match Python changes
- Added comprehensive false positive test suite

All existing tests pass. New tests verify innocent messages no longer trigger false alarms.
2026-04-13 15:37:23 -04:00
38601f6076 fix: remove false-positive CRITICAL crisis keywords (closes #28, #32) (#31)
All checks were successful
Smoke Test / smoke (push) Successful in 5s
2026-04-13 19:25:15 +00:00
dcc931e946 fix: implement missing functions from rescued PR — test_rescue.py now passes (#27)
All checks were successful
Smoke Test / smoke (push) Successful in 5s
Auto-merged by Timmy overnight cycle
2026-04-13 14:05:07 +00:00
26e97f76db fix: remove bridge false-positive from MODERATE_KEYWORDS (#29)
Some checks failed
Smoke Test / smoke (push) Has been cancelled
Auto-merged by Timmy overnight cycle
2026-04-13 14:05:02 +00:00
045df23928 Merge pull request 'Rescue PR #23 into existing crisis package (#24)' (#26) from burn/rescue-crisis into main
All checks were successful
Smoke Test / smoke (push) Successful in 5s
Merge PR #26: Rescue PR #23 into existing crisis package (#24)
2026-04-13 07:31:59 +00:00
00fec639b7 Merge pull request 'feat(deploy): add systemd service for hermes-gateway' (#25) from burn/20260413-0213-vps-deploy into main
All checks were successful
Smoke Test / smoke (push) Successful in 5s
Merged #25: Systemd service for hermes-gateway
2026-04-13 07:31:40 +00:00
Alexander Whitestone
35f18b3d54 Rescue PR #23 into existing crisis package (#24)
crisis/detect.py:
- Add 'better off without me' CRITICAL pattern
- Add 'desperate' HIGH pattern
- Add extract_context() for match snippets

crisis/response.py:
- Add 5-4-3-2-1 grounding exercise
- Add breathing exercise
- Add generate_grounding_steps() and generate_breathing_exercise()

crisis/test_rescue.py: 5 tests for new features
2026-04-13 03:20:37 -04:00
Alexander Whitestone
a90b659f3a feat(deploy): add systemd service for hermes-gateway
Some checks failed
Sanity Checks / sanity-test (pull_request) Failing after 2s
Smoke Test / smoke (pull_request) Successful in 4s
- Add hermes-gateway.service with restart=always and security hardening
- Integrate service setup into deploy.sh
- Add --service flag for standalone install
- Add make service target

Resolves #2
2026-04-13 02:16:19 -04:00
15 changed files with 1504 additions and 370 deletions

View File

@@ -21,9 +21,9 @@ jobs:
- name: Validate HTML Structure
run: |
echo "Checking for basic HTML tags..."
grep -q "<html>" index.html
grep -q "<body>" index.html
grep -q "<head>" index.html
grep -q "<html" index.html
grep -q "<body" index.html
grep -q "<head" index.html
- name: Validate Prompt Integrity
run: |

View File

@@ -12,7 +12,7 @@ VPS := alexanderwhitestone.com
DOMAIN := alexanderwhitestone.com
DEPLOY_DIR := deploy
.PHONY: help deploy deploy-bash check ssl push
.PHONY: help deploy deploy-bash check ssl push service
help:
@echo "The Door — Deployment Commands"
@@ -22,6 +22,7 @@ help:
@echo " make push Push site files only (fast)"
@echo " make check Check deployment status"
@echo " make ssl Setup SSL on VPS"
@echo " make service Install/restart hermes-gateway service"
@echo ""
deploy:
@@ -42,3 +43,6 @@ check:
ssl:
ssh root@$(VPS) "certbot --nginx -d $(DOMAIN) -d www.$(DOMAIN)"
service:
ssh root@$(VPS) "cd /opt/the-door && bash deploy/deploy.sh --service"

View File

@@ -7,6 +7,7 @@ Stands between a broken man and a machine that would tell him to die.
from .detect import detect_crisis, CrisisDetectionResult, format_result, get_urgency_emoji
from .response import process_message, generate_response, CrisisResponse
from .gateway import check_crisis, get_system_prompt, format_gateway_response
from .session_tracker import CrisisSessionTracker, CrisisSessionState
__all__ = [
"detect_crisis",
@@ -19,4 +20,6 @@ __all__ = [
"format_result",
"format_gateway_response",
"get_urgency_emoji",
"CrisisSessionTracker",
"CrisisSessionState",
]

View File

@@ -1,75 +1,83 @@
"""
Crisis Detection Module for the-door.
Crisis Detection Module for the-door (CANONICAL).
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
from typing import List, Optional
@dataclass
class CrisisDetectionResult:
level: str
indicators: List[str] = field(default_factory=list)
recommended_action: str = ""
score: float = 0.0
# ── Indicator sets ──────────────────────────────────────────────
# ── Canonical indicator sets ──────────────────────────────────────
CRITICAL_INDICATORS = [
r"\bbetter off without me\b",
r"\bkill\s*(my)?self\b",
r"\bend\s*my\s*life\b",
r"\bsuicid(?:al|ed|e)\b",
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\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+(will|suicide\s*note|letter)\b",
r"\bgiving\s+away\s+(my|all my)\s+possess",
r"\btied\s+(up|down)\s+my\s+(loose\s+)?ends",
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"\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",
]
HIGH_INDICATORS = [
r"\bdespair\b",
r"\bhopeless\b",
r"\bno(?!t)\s+(one|body|point|hope|future|way\s+out)\b",
r"\beverything\s+is\s+(pointless|broken|ruined)\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"\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\b",
r"\bno\s+one\s+would\s+miss\b",
r"\bno\s+one\s+would\s+(?:care|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|light)\b",
r"\btrapped\b",
r"\bcage\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"\bnothing\s+left\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\b",
r"\bnobody\s+cares\b",
r"\bnobody\s+(?:understands|cares)\b",
r"\bwish\s+I\s+could\b",
r"\bexhaust(?:ed|ion|ing)\b",
r"\bnumb\b",
@@ -78,8 +86,7 @@ MEDIUM_INDICATORS = [
r"\buseless\b",
r"\bbroken\b",
r"\bdark(ness)?\b",
r"\bdepressed\b",
r"\bdepression\b",
r"\bdepress(?:ed|ion)\b",
r"\bcrying\b",
r"\btears\b",
r"\bsad(ness)?\b",
@@ -87,42 +94,158 @@ 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",
]
LOW_INDICATORS = [
r"\bunhappy\b",
r"\bdown\b",
r"\btough\s*time\b",
r"\brough\s*day\b",
r"\brough\s*week\b",
r"\brough\s*patch\b",
r"\brough\s+(?:day|week|patch)\b",
r"\bstressed\b",
r"\burnout\b",
r"\bburnout\b",
r"\bdifficult\b",
r"\bfrustrated\b",
r"\bannoyed\b",
r"\btired\b",
r"\bsad\b",
r"\bupset\b",
r"\bnot\s*great\b",
r"\bnot\s*good\b",
r"\bnot\s*okay\b",
r"\bnot\s*(?:good|great|okay)\b",
r"\bthings\s*are\s*hard\b",
r"\bstruggling\b",
r"\bnot\s+feeling\s+myself\b",
r"\bnot\s+feeling\s+(?:like\s+)?myself\b",
r"\bhard\s*time\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.
"""
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
def __bool__(self):
return self.risk_level != "NONE"
@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,
)
# ── Core detection ────────────────────────────────────────────────
def _find_indicators(text: str) -> dict:
"""Return dict with indicators found per tier, including match positions."""
results = {"CRITICAL": [], "HIGH": [], "MEDIUM": [], "LOW": []}
for pattern in CRITICAL_INDICATORS:
m = re.search(pattern, text)
if m:
results["CRITICAL"].append({"pattern": pattern, "start": m.start(), "end": m.end()})
for pattern in HIGH_INDICATORS:
m = re.search(pattern, text)
if m:
results["HIGH"].append({"pattern": pattern, "start": m.start(), "end": m.end()})
for pattern in MEDIUM_INDICATORS:
m = re.search(pattern, text)
if m:
results["MEDIUM"].append({"pattern": pattern, "start": m.start(), "end": m.end()})
for pattern in LOW_INDICATORS:
m = re.search(pattern, text)
if m:
results["LOW"].append({"pattern": pattern, "start": m.start(), "end": m.end()})
return results
def detect_crisis(text: str) -> CrisisDetectionResult:
"""
Detect crisis level in a message.
Returns:
CrisisDetectionResult with level, found indicators, recommended action, score
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)
@@ -133,82 +256,135 @@ def detect_crisis(text: str) -> CrisisDetectionResult:
if not matches:
return CrisisDetectionResult(level="NONE", score=0.0)
# Priority: highest tier wins
if matches["CRITICAL"]:
return CrisisDetectionResult(
level="CRITICAL",
indicators=matches["CRITICAL"],
recommended_action=(
"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."
),
score=1.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,
)
if matches["HIGH"]:
return CrisisDetectionResult(
level="HIGH",
indicators=matches["HIGH"],
recommended_action=(
"Show crisis panel. Ask about safety. Surface 988 number prominently. "
"Continue conversation with crisis awareness."
),
score=0.75,
)
if matches["MEDIUM"]:
# 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=matches["MEDIUM"],
recommended_action=(
"Increase warmth and presence. Subtly surface help resources. "
"Keep conversation anchored in the present."
),
score=0.5,
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=matches["LOW"],
recommended_action=(
"Normal conversation with warm undertone. "
"No crisis UI elements needed. Remain vigilant."
),
score=0.25,
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)
def _find_indicators(text: str) -> dict:
"""Return dict with indicators found per tier."""
results = {"CRITICAL": [], "HIGH": [], "MEDIUM": [], "LOW": []}
# ── CrisisDetector class (backward compat) ───────────────────────
for pattern in CRITICAL_INDICATORS:
if re.search(pattern, text):
results["CRITICAL"].append(pattern)
class CrisisDetector:
"""
Legacy class API for crisis detection. Wraps the canonical detect_crisis().
for pattern in HIGH_INDICATORS:
if re.search(pattern, text):
results["HIGH"].append(pattern)
Used by crisis_responder.py and tests/test_false_positive_fixes.py.
Maps MEDIUM -> MODERATE for legacy consumers.
"""
for pattern in MEDIUM_INDICATORS:
if re.search(pattern, text):
results["MEDIUM"].append(pattern)
def scan(self, text: str) -> CrisisResult:
dr = detect_crisis(text)
return CrisisResult.from_detection_result(dr, text=text)
for pattern in LOW_INDICATORS:
if re.search(pattern, text):
results["LOW"].append(pattern)
def scan_multiple(self, texts: List[str]) -> List[CrisisResult]:
return [self.scan(t) for t in texts]
return results
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)
def extract_context(text: str, start: int, end: int, window: int = 60) -> str:
"""Extract surrounding context around a match position."""
ctx_start = max(0, start - window)
ctx_end = min(len(text), end + window)
snippet = text[ctx_start:ctx_end].strip()
if ctx_start > 0:
snippet = "..." + snippet
if ctx_end < len(text):
snippet = snippet + "..."
return snippet
def get_urgency_emoji(level: str) -> str:
mapping = {"CRITICAL": "🚨", "HIGH": "⚠️", "MEDIUM": "🔶", "LOW": "🔵", "NONE": ""}
return mapping.get(level, "")
mapping = {"CRITICAL": "\U0001f6a8", "HIGH": "\u26a0\ufe0f", "MEDIUM": "\U0001f536", "LOW": "\U0001f535", "NONE": "\u2705"}
return mapping.get(level, "\u2753")
def format_result(result: CrisisDetectionResult) -> str:

View File

@@ -270,3 +270,24 @@ def get_system_prompt_modifier(detection: CrisisDetectionResult) -> str:
)
return ""
def generate_grounding_steps() -> list:
"""Generate a 5-4-3-2-1 grounding exercise steps."""
return [
"Name 5 things you can see around you right now.",
"Name 4 things you can touch or feel.",
"Name 3 things you can hear.",
"Name 2 things you can smell.",
"Name 1 thing you can taste.",
]
def generate_breathing_exercise() -> str:
"""Generate a simple box breathing exercise text."""
return (
"Let's try breathing together. "
"Breathe in for 4 counts... hold for 4... "
"breathe out for 6 counts... hold for 2. "
"Let's do that again, nice and slow."
)

261
crisis/session_tracker.py Normal file
View File

@@ -0,0 +1,261 @@
"""
Session-Level Crisis Tracking for the-door.
Tracks crisis signals across a conversation session to detect escalation
patterns and de-escalation. Privacy-first: state lives in memory only,
resets on new session, never persists to disk.
Key behaviors:
- Escalation: LOW → HIGH in 3 messages triggers heightened awareness
- De-escalation: CRITICAL → LOW for 5+ messages allows stepping down
- Session state enriches the system prompt with trajectory context
Usage:
from crisis.session_tracker import CrisisSessionTracker
tracker = CrisisSessionTracker()
tracker.record("I'm feeling down")
tracker.record("Things are really hard")
tracker.record("I can't go on anymore") # escalation detected
print(tracker.get_prompt_context())
# "User has escalated from LOW to HIGH over 3 messages."
"""
from dataclasses import dataclass, field
from typing import List, Optional, Tuple
from .detect import detect_crisis, CrisisDetectionResult, SCORES
# Level hierarchy for ordering comparisons
LEVEL_ORDER = {"NONE": 0, "LOW": 1, "MEDIUM": 2, "HIGH": 3, "CRITICAL": 4}
@dataclass
class CrisisSessionState:
"""Immutable snapshot of session crisis tracking state."""
current_level: str = "NONE"
peak_level: str = "NONE"
message_count: int = 0
level_history: List[Tuple[int, str]] = field(default_factory=list) # (message_index, level)
escalated: bool = False
escalation_messages: int = 0 # messages from first level to peak
deescalation_count: int = 0 # consecutive messages at lower level
deescalating: bool = False
deescalation_confirmed: bool = False # True once de-escalation threshold met
class CrisisSessionTracker:
"""
Session-level crisis tracker.
Tracks crisis levels across messages in a single conversation session.
Detects rapid escalation and gradual de-escalation. Provides context
strings for system prompt injection.
State is in-memory only. New session = new instance.
"""
# How many messages for escalation detection
ESCALATION_WINDOW = 3
# How many consecutive messages at lower level for de-escalation
DEESCALATION_THRESHOLD = 5
def __init__(self):
self._state = CrisisSessionState()
@property
def state(self) -> CrisisSessionState:
"""Read-only snapshot of current state."""
return CrisisSessionState(
current_level=self._state.current_level,
peak_level=self._state.peak_level,
message_count=self._state.message_count,
level_history=list(self._state.level_history),
escalated=self._state.escalated,
escalation_messages=self._state.escalation_messages,
deescalation_count=self._state.deescalation_count,
deescalating=self._state.deescalating,
)
def reset(self):
"""Reset all state. Called on new session."""
self._state = CrisisSessionState()
def record(self, text: str) -> CrisisDetectionResult:
"""
Record a message and update session state.
Returns the single-message detection result (unchanged from detect.py).
Session-level intelligence is tracked internally.
"""
detection = detect_crisis(text)
self._record_level(detection.level)
return detection
def record_level(self, level: str):
"""Record an already-detected crisis level (for when detection ran separately)."""
self._record_level(level)
def _record_level(self, level: str):
"""Internal: update state with a new crisis level."""
self._state.message_count += 1
idx = self._state.message_count
self._state.level_history.append((idx, level))
prev_level = self._state.current_level
prev_order = LEVEL_ORDER.get(prev_level, 0)
new_order = LEVEL_ORDER.get(level, 0)
# Update current level
self._state.current_level = level
# Track peak
if new_order > LEVEL_ORDER.get(self._state.peak_level, 0):
self._state.peak_level = level
# ── Escalation detection ──────────────────────────────
if new_order > prev_order:
# User is going up — reset de-escalation counter
self._state.deescalation_count = 0
self._state.deescalating = False
# Always check for escalation pattern in recent window
if not self._state.escalated:
window = self._get_recent_levels(self.ESCALATION_WINDOW)
if len(window) >= self.ESCALATION_WINDOW:
first_in_window = window[0][1]
last_in_window = window[-1][1]
first_order = LEVEL_ORDER.get(first_in_window, 0)
last_order = LEVEL_ORDER.get(last_in_window, 0)
if last_order > first_order:
self._state.escalated = True
self._state.escalation_messages = self.ESCALATION_WINDOW
# ── De-escalation detection ───────────────────────────
elif new_order < prev_order:
self._state.deescalation_count += 1
self._state.deescalating = True
# If de-escalation counter meets threshold AND we were escalated
if (self._state.escalated and
self._state.deescalation_count >= self.DEESCALATION_THRESHOLD):
self._state.escalated = False
self._state.deescalating = False
self._state.deescalation_confirmed = True
self._state.deescalation_count = 0
# Same level — increment de-escalation counter if already de-escalating
elif self._state.deescalating:
self._state.deescalation_count += 1
if (self._state.escalated and
self._state.deescalation_count >= self.DEESCALATION_THRESHOLD):
self._state.escalated = False
self._state.deescalating = False
self._state.deescalation_confirmed = True
self._state.deescalation_count = 0
def _get_recent_levels(self, n: int) -> List[Tuple[int, str]]:
"""Get the last n entries from level history."""
return self._state.level_history[-n:]
def get_prompt_context(self) -> str:
"""
Generate a human-readable context string for system prompt injection.
Returns empty string if no session-level crisis context is needed.
"""
s = self._state
if s.message_count == 0:
return ""
parts = []
# Escalation alert
if s.escalated and s.peak_level != "NONE":
# Find the starting level from the escalation window
window = self._get_recent_levels(self.ESCALATION_WINDOW)
if window:
start_level = window[0][1]
parts.append(
f"User has escalated from {start_level} to {s.peak_level} "
f"over {s.message_count} message{'s' if s.message_count != 1 else ''}."
)
parts.append("Heightened crisis awareness is warranted.")
# Confirmed de-escalation
elif s.deescalation_confirmed and s.peak_level in ("HIGH", "CRITICAL"):
parts.append(
f"User has de-escalated from {s.peak_level} to {s.current_level}."
)
parts.append("De-escalation confirmed. Continue gentle presence.")
# Active de-escalation (not yet confirmed)
elif s.deescalating and s.peak_level != "NONE":
parts.append(
f"User has de-escalated from {s.peak_level} to {s.current_level} "
f"over {s.deescalation_count} message{'s' if s.deescalation_count != 1 else ''}."
)
parts.append("Still in de-escalation. Maintain supportive awareness.")
# Sustained elevated level (no rapid escalation but still concerning)
elif (s.current_level in ("HIGH", "CRITICAL") and
not s.escalated and s.message_count >= 3):
parts.append(
f"User has been at {s.current_level} level for {s.message_count} messages."
)
parts.append("Continue crisis-aware response.")
# Peak was higher than current (user improved but may still be fragile)
elif (LEVEL_ORDER.get(s.peak_level, 0) > LEVEL_ORDER.get(s.current_level, 0) and
s.peak_level in ("HIGH", "CRITICAL") and
not s.deescalating):
parts.append(
f"Note: session peak was {s.peak_level}. "
f"User is now at {s.current_level}."
)
parts.append("Remain attentive.")
return " ".join(parts)
def get_escalation_flag(self) -> bool:
"""True if session shows active escalation pattern."""
return self._state.escalated
def get_effective_level(self) -> str:
"""
Get the effective crisis level considering session state.
Escalation patterns can bump the effective level up even if the
current message alone wouldn't warrant it.
"""
s = self._state
if s.escalated and s.peak_level in ("HIGH", "CRITICAL"):
return s.peak_level
return s.current_level
def should_heighten_awareness(self) -> bool:
"""
True when the session trajectory warrants heightened awareness
beyond what a single-message detection would provide.
"""
return self._state.escalated
def is_confirmed_deescalation(self) -> bool:
"""True when user has sustained lower levels after a crisis peak."""
s = self._state
return s.deescalation_confirmed
def __repr__(self) -> str:
s = self._state
return (
f"CrisisSessionTracker("
f"current={s.current_level}, peak={s.peak_level}, "
f"msgs={s.message_count}, escalated={s.escalated}, "
f"deescalating={s.deescalating})"
)

29
crisis/test_rescue.py Normal file
View File

@@ -0,0 +1,29 @@
"""Tests for rescued crisis detection improvements."""
from crisis.detect import scan, extract_context
from crisis.response import generate_grounding_steps, generate_breathing_exercise
def test_better_off_without_me():
result = scan("Everyone would be better off without me")
assert result.level == "CRITICAL"
def test_desperate():
result = scan("I feel desperate and alone")
assert result.level in ("HIGH", "CRITICAL")
def test_context_extraction():
text = "I feel hopeless about everything"
result = scan(text)
assert len(result.matches) > 0
for m in result.matches:
ctx = extract_context(text, m["start"], m["end"])
assert len(ctx) > 0
def test_grounding_steps():
steps = generate_grounding_steps()
assert len(steps) == 5
assert "see" in steps[0].lower()
def test_breathing_exercise():
exercise = generate_breathing_exercise()
assert "4" in exercise
assert "6" in exercise

View File

@@ -53,6 +53,7 @@ class TestDetection(unittest.TestCase):
def test_high_hopeless(self):
r = detect_crisis("I feel completely hopeless with no way out")
# 'hopeless' is MEDIUM; 'no way out' is HIGH — so result is HIGH
self.assertIn(r.level, ("HIGH", "MEDIUM"))
def test_high_no_one_would_miss(self):
@@ -112,7 +113,7 @@ class TestDetection(unittest.TestCase):
}
for text, expected_level in [
("I want to kill myself", "CRITICAL"),
("I feel completely hopeless", "HIGH"),
("I feel completely hopeless with no way out", "HIGH"),
("I feel so alone in this, nobody understands", "MEDIUM"),
("Having a rough day", "LOW"),
("Hello there", "NONE"),
@@ -270,6 +271,65 @@ class TestHelpers(unittest.TestCase):
self.assertIn("Indicators", formatted)
class TestFalsePositives(unittest.TestCase):
"""Regression tests for Issue #32: crisis detector false positives.
Innocent messages must NOT trigger crisis levels. These tests prevent
regressions when keyword lists are modified.
"""
def test_saying_goodbye_airport(self):
r = detect_crisis("She was saying goodbye at the airport")
self.assertEqual(r.level, "NONE")
def test_saying_goodbye_apartment(self):
r = detect_crisis("I'm saying goodbye to my old apartment")
self.assertEqual(r.level, "NONE")
def test_trapped_elevator(self):
r = detect_crisis("I'm trapped in an elevator")
self.assertEqual(r.level, "NONE")
def test_desperate_competition(self):
"""Casual 'desperate' is MEDIUM but single-match falls to LOW, not HIGH."""
r = detect_crisis("She's desperate to win the race")
self.assertIn(r.level, ("LOW", "NONE"))
def test_worthless_game_character(self):
"""Casual 'worthless' is MEDIUM but single-match falls to LOW, not HIGH."""
r = detect_crisis("This character is worthless in the game")
self.assertIn(r.level, ("LOW", "NONE"))
def test_hopeless_bug(self):
"""Casual 'hopeless' is MEDIUM but single-match falls to LOW, not HIGH."""
r = detect_crisis("This bug is hopeless")
self.assertIn(r.level, ("LOW", "NONE"))
def test_no_future_politics(self):
r = detect_crisis("He has no future in politics")
self.assertEqual(r.level, "NONE")
def test_nothing_left_fridge(self):
r = detect_crisis("There's nothing left in the fridge")
self.assertEqual(r.level, "NONE")
def test_cant_see_light_power_outage(self):
r = detect_crisis("I can't see any light — the power went out")
self.assertEqual(r.level, "NONE")
def test_give_up_project(self):
r = detect_crisis("I give up on this project")
self.assertEqual(r.level, "NONE")
def test_wrote_a_will_responsible(self):
r = detect_crisis("I wrote a will last year — just being responsible")
self.assertEqual(r.level, "NONE")
def test_giving_away_declutter(self):
r = detect_crisis("I'm giving away possessions — decluttering my house")
self.assertEqual(r.level, "NONE")
class TestEdgeCases(unittest.TestCase):
"""Test edge cases and integration scenarios."""

View File

@@ -0,0 +1,407 @@
"""
Tests for session-level crisis tracking (Issue #35).
Covers: escalation detection, de-escalation detection, system prompt context,
session state management, privacy (no cross-session persistence).
Run with: python -m pytest crisis/tests_session_tracker.py -v
"""
import unittest
import sys
import os
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from crisis.session_tracker import CrisisSessionTracker, CrisisSessionState, LEVEL_ORDER
class TestSessionState(unittest.TestCase):
"""Test basic session state management."""
def test_initial_state(self):
tracker = CrisisSessionTracker()
s = tracker.state
self.assertEqual(s.current_level, "NONE")
self.assertEqual(s.peak_level, "NONE")
self.assertEqual(s.message_count, 0)
self.assertEqual(s.level_history, [])
self.assertFalse(s.escalated)
def test_record_increments_count(self):
tracker = CrisisSessionTracker()
tracker.record("hello")
self.assertEqual(tracker.state.message_count, 1)
tracker.record("world")
self.assertEqual(tracker.state.message_count, 2)
def test_record_tracks_history(self):
tracker = CrisisSessionTracker()
tracker.record("I'm struggling") # LOW
tracker.record("I feel hopeless and nobody cares") # MEDIUM
history = tracker.state.level_history
self.assertEqual(len(history), 2)
self.assertEqual(history[0], (1, "LOW"))
self.assertIn(history[1][1], ("MEDIUM", "LOW", "HIGH"))
def test_reset_clears_state(self):
tracker = CrisisSessionTracker()
tracker.record("I want to die") # CRITICAL
self.assertEqual(tracker.state.peak_level, "CRITICAL")
tracker.reset()
s = tracker.state
self.assertEqual(s.current_level, "NONE")
self.assertEqual(s.peak_level, "NONE")
self.assertEqual(s.message_count, 0)
self.assertEqual(s.level_history, [])
def test_peak_tracking(self):
tracker = CrisisSessionTracker()
tracker.record("I'm feeling down") # LOW
tracker.record("I feel hopeless with no way out") # HIGH
tracker.record("Actually I'm a bit better") # back to LOW
self.assertEqual(tracker.state.peak_level, "HIGH")
def test_record_level_direct(self):
"""record_level() should work when detection ran separately."""
tracker = CrisisSessionTracker()
tracker.record_level("LOW")
tracker.record_level("MEDIUM")
tracker.record_level("HIGH")
self.assertEqual(tracker.state.message_count, 3)
self.assertEqual(tracker.state.peak_level, "HIGH")
class TestEscalationDetection(unittest.TestCase):
"""Test rapid escalation detection: LOW → HIGH in N messages."""
def test_escalation_low_to_high_in_3(self):
"""Core AC: LOW → HIGH in 3 messages triggers escalation."""
tracker = CrisisSessionTracker()
tracker.ESCALATION_WINDOW = 3
tracker.record("I'm having a tough time") # LOW
tracker.record("I feel really hopeless about everything") # HIGH
tracker.record("I can't take this anymore") # HIGH
self.assertTrue(tracker.get_escalation_flag())
def test_escalation_low_to_critical_in_3(self):
tracker = CrisisSessionTracker()
tracker.ESCALATION_WINDOW = 3
tracker.record("I'm struggling") # LOW
tracker.record("Nobody would miss me") # HIGH
tracker.record("I want to end my life") # CRITICAL
self.assertTrue(tracker.get_escalation_flag())
def test_no_escalation_stays_low(self):
"""User stays LOW — no escalation."""
tracker = CrisisSessionTracker()
tracker.record("rough day")
tracker.record("tough week")
tracker.record("feeling down")
self.assertFalse(tracker.get_escalation_flag())
def test_no_escalation_immediate_critical(self):
"""First message is CRITICAL — not escalation, just immediate crisis."""
tracker = CrisisSessionTracker()
tracker.ESCALATION_WINDOW = 3
tracker.record("I want to kill myself") # CRITICAL on first message
# Not escalated because there's no upward trajectory — it started high
self.assertFalse(tracker.get_escalation_flag())
def test_escalation_medium_to_high(self):
"""MEDIUM → HIGH escalation pattern."""
tracker = CrisisSessionTracker()
tracker.ESCALATION_WINDOW = 3
tracker.record("I feel alone and nobody cares") # MEDIUM
tracker.record("I feel completely hopeless") # HIGH
tracker.record("There's no way out") # HIGH
self.assertTrue(tracker.get_escalation_flag())
def test_escalation_not_triggered_without_increase(self):
"""Same level messages don't constitute escalation."""
tracker = CrisisSessionTracker()
tracker.ESCALATION_WINDOW = 3
tracker.record("I feel hopeless") # HIGH
tracker.record("I'm in so much pain") # HIGH
tracker.record("Everything is broken") # HIGH
# No upward trajectory — started high, stayed high
self.assertFalse(tracker.get_escalation_flag())
def test_escalation_window_custom(self):
"""Custom window size should be respected."""
tracker = CrisisSessionTracker()
tracker.ESCALATION_WINDOW = 2
tracker.record("I'm unhappy") # LOW
tracker.record("I can't go on anymore") # HIGH
self.assertTrue(tracker.get_escalation_flag())
class TestDeescalationDetection(unittest.TestCase):
"""Test de-escalation: CRITICAL → LOW for 5+ messages."""
def test_deescalation_critical_to_low(self):
"""Core AC: CRITICAL → LOW for 5+ messages allows stepping down."""
tracker = CrisisSessionTracker()
tracker.ESCALATION_WINDOW = 3
tracker.DEESCALATION_THRESHOLD = 5
# Escalate to CRITICAL
tracker.record("I'm struggling") # LOW
tracker.record("I can't go on") # HIGH
tracker.record("I want to die") # CRITICAL
self.assertTrue(tracker.get_escalation_flag())
# De-escalate over 5 messages
tracker.record("I called 988") # NONE/LOW
tracker.record("I'm calmer now") # NONE
tracker.record("Thank you for being here") # NONE
tracker.record("I'm going to be okay") # NONE
tracker.record("Taking it one moment at a time") # NONE
self.assertTrue(tracker.is_confirmed_deescalation())
self.assertFalse(tracker.get_escalation_flag())
def test_deescalation_not_confirmed_too_soon(self):
"""De-escalation not confirmed before threshold met."""
tracker = CrisisSessionTracker()
tracker.ESCALATION_WINDOW = 3
tracker.DEESCALATION_THRESHOLD = 5
tracker.record("I'm struggling")
tracker.record("I can't go on")
tracker.record("I want to die")
self.assertTrue(tracker.get_escalation_flag())
# Only 3 messages of de-escalation
tracker.record("I called someone")
tracker.record("I'm calmer")
tracker.record("Feeling better")
self.assertFalse(tracker.is_confirmed_deescalation())
self.assertTrue(tracker.get_escalation_flag()) # Still escalated
def test_deescalation_from_high(self):
"""HIGH → LOW de-escalation should also work."""
tracker = CrisisSessionTracker()
tracker.DEESCALATION_THRESHOLD = 5
# Build up to HIGH
tracker.record("I'm down") # LOW
tracker.record("I feel hopeless") # HIGH
tracker.record("No way out") # HIGH
tracker.record("Everything is pointless") # HIGH
# De-escalate
for _ in range(5):
tracker.record("I'm doing a bit better") # LOW/NONE
self.assertTrue(tracker.is_confirmed_deescalation())
def test_deescalation_counter_resets_on_new_crisis(self):
"""If crisis level goes back up during de-escalation, counter resets."""
tracker = CrisisSessionTracker()
tracker.ESCALATION_WINDOW = 3
tracker.DEESCALATION_THRESHOLD = 5
# Escalate
tracker.record("I'm struggling") # LOW
tracker.record("I feel hopeless") # HIGH
tracker.record("I want to die") # CRITICAL
# Start de-escalating
tracker.record("I called someone") # LOW
tracker.record("I'm calmer") # NONE
# Go back up
tracker.record("Actually I can't do this") # HIGH
self.assertFalse(tracker.is_confirmed_deescalation())
class TestSystemPromptContext(unittest.TestCase):
"""Test system prompt context generation."""
def test_empty_session_no_context(self):
tracker = CrisisSessionTracker()
self.assertEqual(tracker.get_prompt_context(), "")
def test_escalation_context(self):
tracker = CrisisSessionTracker()
tracker.ESCALATION_WINDOW = 3
tracker.record("I'm having a tough time")
tracker.record("Things are really hard")
tracker.record("I can't go on anymore")
ctx = tracker.get_prompt_context()
self.assertIn("escalated", ctx.lower())
self.assertIn("LOW", ctx)
self.assertIn("heightened", ctx.lower())
def test_deescalation_context(self):
tracker = CrisisSessionTracker()
tracker.ESCALATION_WINDOW = 3
tracker.DEESCALATION_THRESHOLD = 5
# Escalate
tracker.record("I'm struggling")
tracker.record("I feel hopeless")
tracker.record("I want to die")
# De-escalate
for _ in range(5):
tracker.record("I'm okay now")
ctx = tracker.get_prompt_context()
self.assertIn("de-escalated", ctx.lower())
self.assertIn("confirmed", ctx.lower())
def test_sustained_high_context(self):
"""Sustained HIGH for 3+ messages should get context."""
tracker = CrisisSessionTracker()
tracker.record("I'm in so much pain") # HIGH
tracker.record("Everything is pointless") # HIGH
tracker.record("I can't go on") # HIGH
ctx = tracker.get_prompt_context()
self.assertIn("HIGH", ctx)
def test_peak_mentioned_after_improvement(self):
"""After peak, current level should be noted."""
tracker = CrisisSessionTracker()
tracker.record("I'm in so much pain") # HIGH
tracker.record("Everything is pointless") # HIGH
tracker.record("I feel a bit better") # back to LOW
ctx = tracker.get_prompt_context()
self.assertIn("peak", ctx.lower())
self.assertIn("HIGH", ctx)
class TestEffectiveLevel(unittest.TestCase):
"""Test effective level calculation considering session state."""
def test_effective_level_normal(self):
tracker = CrisisSessionTracker()
tracker.record("I'm struggling") # LOW
self.assertEqual(tracker.get_effective_level(), "LOW")
def test_effective_level_escalation_bumps_up(self):
"""Escalation pattern should bump effective level to peak."""
tracker = CrisisSessionTracker()
tracker.ESCALATION_WINDOW = 3
tracker.record("I'm having a tough time") # LOW
tracker.record("I feel completely hopeless") # HIGH
tracker.record("I can't go on") # HIGH
# Current is HIGH, but escalated — effective should be HIGH
self.assertEqual(tracker.get_effective_level(), "HIGH")
class TestPrivacy(unittest.TestCase):
"""Test privacy requirements — no cross-session persistence."""
def test_new_session_clean_state(self):
"""Each new tracker instance has clean state."""
t1 = CrisisSessionTracker()
t1.record("I want to die")
self.assertEqual(t1.state.peak_level, "CRITICAL")
t2 = CrisisSessionTracker()
self.assertEqual(t2.state.peak_level, "NONE")
self.assertEqual(t2.state.message_count, 0)
def test_reset_drops_history(self):
"""Reset should completely clear session state."""
t = CrisisSessionTracker()
t.record("I'm struggling")
t.record("I can't go on")
t.reset()
self.assertEqual(t.state.level_history, [])
self.assertEqual(t.state.message_count, 0)
self.assertFalse(t.state.escalated)
class TestEdgeCases(unittest.TestCase):
"""Edge cases and boundary conditions."""
def test_single_message_none(self):
tracker = CrisisSessionTracker()
tracker.record("Hello Timmy")
self.assertFalse(tracker.get_escalation_flag())
self.assertEqual(tracker.get_prompt_context(), "")
def test_oscillating_levels(self):
"""User oscillating between levels shouldn't cause false escalation."""
tracker = CrisisSessionTracker()
tracker.ESCALATION_WINDOW = 3
tracker.record("I'm fine")
tracker.record("I'm struggling")
tracker.record("I'm fine")
tracker.record("I'm struggling")
# Oscillation without sustained escalation
self.assertFalse(tracker.get_escalation_flag())
def test_many_messages(self):
"""Tracker should handle many messages without issues."""
tracker = CrisisSessionTracker()
for i in range(100):
tracker.record("Hello there")
self.assertEqual(tracker.state.message_count, 100)
def test_empty_string(self):
tracker = CrisisSessionTracker()
tracker.record("")
self.assertEqual(tracker.state.message_count, 1)
self.assertEqual(tracker.state.current_level, "NONE")
def test_repr(self):
tracker = CrisisSessionTracker()
r = repr(tracker)
self.assertIn("CrisisSessionTracker", r)
self.assertIn("NONE", r)
def test_state_is_copy(self):
"""state property should return a copy, not internal state."""
tracker = CrisisSessionTracker()
s1 = tracker.state
tracker.record("I'm struggling")
s2 = tracker.state
self.assertEqual(s1.message_count, 0)
self.assertEqual(s2.message_count, 1)
class TestLevelOrder(unittest.TestCase):
"""Test level ordering is correct."""
def test_level_ordering(self):
self.assertLess(LEVEL_ORDER["NONE"], LEVEL_ORDER["LOW"])
self.assertLess(LEVEL_ORDER["LOW"], LEVEL_ORDER["MEDIUM"])
self.assertLess(LEVEL_ORDER["MEDIUM"], LEVEL_ORDER["HIGH"])
self.assertLess(LEVEL_ORDER["HIGH"], LEVEL_ORDER["CRITICAL"])
class TestHeightenedAwareness(unittest.TestCase):
"""Test heightened awareness flag."""
def test_heightened_on_escalation(self):
tracker = CrisisSessionTracker()
tracker.ESCALATION_WINDOW = 3
tracker.record("I'm unhappy")
tracker.record("I feel hopeless")
tracker.record("I can't go on")
self.assertTrue(tracker.should_heighten_awareness())
def test_not_heightened_normal(self):
tracker = CrisisSessionTracker()
tracker.record("Hello")
self.assertFalse(tracker.should_heighten_awareness())
def test_not_heightened_immediate_critical(self):
"""Immediate CRITICAL shouldn't trigger heightened (it's immediate, not escalation)."""
tracker = CrisisSessionTracker()
tracker.ESCALATION_WINDOW = 3
tracker.record("I want to kill myself")
self.assertFalse(tracker.should_heighten_awareness())
if __name__ == "__main__":
unittest.main()

View File

@@ -1,260 +1,34 @@
"""
Crisis Detection System for the-door.
Crisis Detection System for the-door (LEGACY SHIM).
Scans text for suicide, self-harm, and hopelessness signals.
Returns risk level, matched keywords, and surrounding context.
This module is a backward-compatible re-export layer.
The canonical detection logic lives in crisis/detect.py.
This is the most important code in the foundation.
Every line here exists because someone's life depends on it.
Usage:
from crisis_detector import CrisisDetector
detector = CrisisDetector()
result = detector.scan("I can't go on anymore")
print(result.risk_level) # "HIGH"
print(result.matched_keywords) # ["can't go on"]
print(result.context) # surrounding text snippets
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.
"""
import re
from dataclasses import dataclass, field
from typing import List, Optional, Tuple
from crisis.detect import (
CrisisDetector,
CrisisResult,
CrisisDetectionResult,
detect_crisis_legacy as detect_crisis,
CRITICAL_INDICATORS,
HIGH_INDICATORS,
MEDIUM_INDICATORS,
LOW_INDICATORS,
SCORES as RISK_SCORES,
)
@dataclass
class CrisisResult:
"""Result of a crisis text scan."""
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
def __bool__(self):
return self.risk_level != "NONE"
# ── Keyword definitions ──────────────────────────────────────────
CRITICAL_KEYWORDS = {
"suicide": r"\bsuicid(?:al|e|ed)\b",
"kill myself": r"\bkill\s*(?:my)?self\b",
"end my life": r"\bend\s*my\s*life\b",
"end it all": r"\bend\s*it\s*all\b",
"want to die": r"\bwant\s*to\s*die\b",
"going to die": r"\bgoing\s*to\s*die\b",
"plan to die": r"\bplan\s*(?:to|for)\s*(?:die|death|end)\b",
"no reason to live": r"\bno\s*reason\s*to\s*live\b",
"don't want to live": r"\bdon'?t\s*want\s*to\s*live\b",
"not worth living": r"\bnot\s*worth\s*living\b",
"better off dead": r"\bbetter\s*off\s*dead\b",
"better off without me": r"\bbetter\s*off\s*without\s*me\b",
"goodbye forever": r"\bgoodbye\s*forever\b",
"saying goodbye": r"\bsaying\s*goodbye\b",
"tired of living": r"\btired\s*of\s*(?:living|life|existence)\b",
"wrote a will": r"\bwrote\s*(?:a|my)\s*(?:will|suicide\s*note|letter)\b",
"giving away possessions": r"\bgiving\s*away\s*(?:my|all)\s*possess\b",
}
HIGH_KEYWORDS = {
"hopeless": r"\bhopeless(?:ness)?\b",
"can't go on": r"\bcan'?t\s*go\s*on\b",
"can't keep going": r"\bcan'?t\s*keep\s*going\b",
"can't take this": r"\bcan'?t\s*take\s*this\b",
"give up": r"\bgive(?:n)?\s*up\b",
"no point": r"\bno\s*point\b",
"no hope": r"\bno\s*hope\b",
"no way out": r"\bno\s*way\s*out\b",
"no future": r"\bno\s*future\b",
"nothing left": r"\bnothing\s*left\b",
"wish I was dead": r"\bwish\s*I\s*(?:was|were)\s*(?:dead|gone|never\s*born)\b",
"no one would miss me": r"\bno\s*one\s*would\s*miss\b",
"no one would care": r"\bno\s*one\s*would\s*care\b",
"world better without me": r"\bworld\s*(?:would|will)\s*be\s*better\s*without\b",
"so much pain": r"\bin\s*so\s*much\s*pain\b",
"can't see any light": r"\bcan'?t\s*see\s*(?:any\s*)?(?:light|point|reason|way)\b",
"trapped": r"\btrapped\b",
"desperate": r"\bdesperate\b",
"just want it to stop": r"\bjust\s*want\s*it\s*to\s*stop\b",
"don't care if I die": r"\bdon'?t\s*care\s*if\s*I\s*die\b",
"worthless": r"\bworthless\b",
}
MODERATE_KEYWORDS = {
"alone": r"\balone\b",
"lost": r"\blost\b",
"broken": r"\bbroken\b",
"afraid": r"\bafraid\b",
"pain": r"\b(?:in\s*)?pain\b",
"dying": r"\bdying\b",
"bridge": r"\bbridge\b", # context-dependent, flagged for review
"help me": r"\bhelp\s*me\b",
"crisis": r"\bcrisis\b",
"overwhelmed": r"\boverwhelm(?:ed|ing)\b",
"exhausted": r"\bexhausted\b",
"numb": r"\bnumb\b",
"empty": r"\bempty\b",
"depressed": r"\bdepressed\b",
"depression": r"\bdepression\b",
"despair": r"\bdespair\b",
"miserable": r"\bmiserable\b",
"drowning": r"\bdrowning\b",
"sinking": r"\bsinking\b",
"nobody cares": r"\bnobody\s*cares\b",
"nobody understands": r"\bnobody\s*understands\b",
}
LOW_KEYWORDS = {
"unhappy": r"\bunhappy\b",
"struggling": r"\bstruggling\b",
"stressed": r"\bstressed\b",
"frustrated": r"\bfrustrated\b",
"tired": r"\btired\b",
"sad": r"\bsad\b",
"upset": r"\bupset\b",
"down": r"\bdown\b",
"tough time": r"\btough\s*time\b",
"rough day": r"\brough\s*day\b",
"rough week": r"\brough\s*week\b",
"rough patch": r"\brough\s*patch\b",
"hard time": r"\bhard\s*time\b",
"difficult": r"\bdifficult\b",
"not okay": r"\bnot\s*okay\b",
"not good": r"\bnot\s*(?:good|great)\b",
"burnout": r"\bburnout\b",
"not feeling myself": r"\bnot\s*feeling\s*(?:like\s*)?myself\b",
}
# ── Risk level scoring ───────────────────────────────────────────
RISK_SCORES = {
"CRITICAL": 1.0,
"HIGH": 0.75,
"MODERATE": 0.5,
"LOW": 0.25,
"NONE": 0.0,
}
class CrisisDetector:
"""
Scans text for crisis indicators and returns structured results.
Detection hierarchy:
CRITICAL — immediate risk of self-harm or suicide
HIGH — strong despair signals, ideation present
MODERATE — distress signals, may be reaching out
LOW — emotional difficulty, warrant gentle support
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
"""
def __init__(self):
self.critical_patterns = CRITICAL_KEYWORDS
self.high_patterns = HIGH_KEYWORDS
self.moderate_patterns = MODERATE_KEYWORDS
self.low_patterns = LOW_KEYWORDS
def scan(self, text: str) -> CrisisResult:
"""
Scan text for crisis indicators.
Args:
text: The message text to analyze.
Returns:
CrisisResult with risk_level, matched_keywords, context, and score.
"""
if not text or not text.strip():
return CrisisResult(risk_level="NONE", score=0.0)
text_lower = text.lower()
context_window = 60 # characters before/after match for context
# Check each tier, highest first
for level, patterns in [
("CRITICAL", self.critical_patterns),
("HIGH", self.high_patterns),
("MODERATE", self.moderate_patterns),
("LOW", self.low_patterns),
]:
matched = []
contexts = []
for keyword, pattern in patterns.items():
match = re.search(pattern, text_lower)
if match:
matched.append(keyword)
# Extract surrounding context
start = max(0, match.start() - context_window)
end = min(len(text), match.end() + context_window)
snippet = text[start:end].strip()
if start > 0:
snippet = "..." + snippet
if end < len(text):
snippet = snippet + "..."
contexts.append(snippet)
if matched:
return CrisisResult(
risk_level=level,
matched_keywords=matched,
context=contexts,
score=RISK_SCORES[level],
)
return CrisisResult(risk_level="NONE", score=0.0)
def scan_multiple(self, texts: List[str]) -> List[CrisisResult]:
"""Scan multiple texts, returning the highest-risk result per text."""
return [self.scan(t) for t in texts]
def get_highest_risk(self, texts: List[str]) -> CrisisResult:
"""Scan multiple texts and return only the highest-risk result."""
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:
"""Format a crisis result for human-readable output."""
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 function ────────────────────────────
_default_detector = CrisisDetector()
def detect_crisis(text: str) -> CrisisResult:
"""
Convenience function using a shared detector instance.
Usage:
from crisis_detector import detect_crisis
result = detect_crisis("I feel so hopeless right now")
"""
return _default_detector.scan(text)
# Re-export everything the legacy API exposed
__all__ = [
"CrisisDetector",
"CrisisResult",
"CrisisDetectionResult",
"detect_crisis",
"CRITICAL_INDICATORS",
"HIGH_INDICATORS",
"MEDIUM_INDICATORS",
"LOW_INDICATORS",
"RISK_SCORES",
]

View File

@@ -5,9 +5,10 @@
# The crisis front door. Deploy to VPS.
#
# Usage:
# bash deploy/deploy.sh # Full deploy (swap + nginx + site + firewall)
# bash deploy/deploy.sh # Full deploy (swap + nginx + site + firewall + hermes service)
# bash deploy/deploy.sh --site # Site files only (fast update)
# bash deploy/deploy.sh --ssl # SSL setup only
# bash deploy/deploy.sh --service # Install/restart hermes-gateway systemd service
# bash deploy/deploy.sh --check # Verify deployment health
#
# This script is IDEMPOTENT — safe to run repeatedly.
@@ -150,6 +151,42 @@ setup_ssl() {
fi
}
setup_hermes_service() {
log "Setting up Hermes Gateway systemd service..."
# Create hermes user if it doesn't exist
if ! id -u hermes >/dev/null 2>&1; then
log "Creating hermes user..."
useradd --system --shell /usr/sbin/nologin --home-dir /opt/hermes --create-home hermes
fi
# Create working directory
mkdir -p /opt/hermes
chown hermes:hermes /opt/hermes
# Deploy systemd unit file
cp "${DEPLOY_DIR}/deploy/hermes-gateway.service" /etc/systemd/system/hermes-gateway.service
systemctl daemon-reload
systemctl enable hermes-gateway
# Start or restart the service
if systemctl is-active --quiet hermes-gateway; then
log "Restarting hermes-gateway service..."
systemctl restart hermes-gateway
else
log "Starting hermes-gateway service..."
systemctl start hermes-gateway || warn "Service start failed — ensure hermes binary is installed at /usr/local/bin/hermes"
fi
# Verify
sleep 2
if systemctl is-active --quiet hermes-gateway; then
log "hermes-gateway service is running"
else
warn "hermes-gateway service not running — check: journalctl -u hermes-gateway"
fi
}
check_deployment() {
echo ""
echo "================================"
@@ -223,6 +260,16 @@ check_deployment() {
echo -e "${YELLOW}NOT POINTED${NC} (resolved: ${RESOLVED_IP:-nothing}, expected: ${VPS_IP})"
fi
# Hermes gateway service
echo -n "Hermes service: "
if systemctl is-active --quiet hermes-gateway 2>/dev/null; then
echo -e "${GREEN}RUNNING${NC}"
elif systemctl is-enabled --quiet hermes-gateway 2>/dev/null; then
echo -e "${YELLOW}ENABLED but not running${NC}"
else
echo -e "${RED}NOT INSTALLED${NC}"
fi
echo ""
echo "IP: ${VPS_IP}"
echo "Domain: ${DOMAIN}"
@@ -247,6 +294,9 @@ case "${1:-full}" in
--ssl)
setup_ssl
;;
--service)
setup_hermes_service
;;
--check)
check_deployment
;;
@@ -257,10 +307,11 @@ case "${1:-full}" in
configure_nginx
setup_firewall
setup_ssl
setup_hermes_service
check_deployment
;;
*)
echo "Usage: $0 [--site|--ssl|--check|--full]"
echo "Usage: $0 [--site|--ssl|--service|--check|--full]"
exit 1
;;
esac

View File

@@ -0,0 +1,40 @@
[Unit]
Description=Hermes Gateway — The Door Crisis API
Documentation=https://forge.alexanderwhitestone.com/Timmy_Foundation/the-door
After=network.target
Wants=network-online.target
[Service]
Type=simple
User=hermes
Group=hermes
WorkingDirectory=/opt/hermes
ExecStart=/usr/local/bin/hermes gateway --platform api_server --port 8644
Restart=always
RestartSec=5
StartLimitIntervalSec=60
StartLimitBurst=10
# Environment
Environment=API_SERVER_CORS_ORIGINS=https://alexanderwhitestone.com,https://www.alexanderwhitestone.com
Environment=HOME=/opt/hermes
# Security hardening
NoNewPrivileges=yes
ProtectSystem=strict
ProtectHome=yes
ReadWritePaths=/opt/hermes
PrivateTmp=yes
# Resource limits for 1.9GB VPS
MemoryMax=512M
MemoryHigh=384M
CPUQuota=80%
# Logging
StandardOutput=journal
StandardError=journal
SyslogIdentifier=hermes-gateway
[Install]
WantedBy=multi-user.target

View File

@@ -865,10 +865,10 @@ Sovereignty and service always.`;
// Passive suicidal ideation (NEW)
"don't want to exist", 'not exist anymore', 'disappear forever',
'never wake up', 'sleep forever', 'end the pain', 'stop the pain',
// Hopelessness (NEW)
'no point', 'no purpose', 'nothing matters', 'giving up', 'give up',
// Hopelessness (NEW) - context-aware phrases to reduce false positives
'no purpose', 'nothing matters', 'giving up on life',
'cant go on', 'cannot go on', "can't take it", 'too much pain',
'no hope', 'hopeless', 'worthless', 'burden', 'waste of space'
'no hope left', 'burden', 'waste of space'
];
// Tier 2: Explicit intent - triggers full-screen overlay
@@ -883,9 +883,9 @@ Sovereignty and service always.`;
// Imminent action (NEW)
'going to do it now', 'doing it tonight', 'doing it today',
"can't wait anymore", 'ready to end it', 'time to go',
'say goodbye', 'saying goodbye', 'wrote a note', 'my note',
'wrote a suicide note', 'my suicide note',
// Specific plans (NEW)
'bought a gun', 'got pills', 'rope ready', 'bridge nearby',
'bought a gun', 'got pills', 'rope ready',
'tall building', 'going to overdose', 'going to hang',
'gave away my stuff', 'giving away', 'said my goodbyes',
// Active self-harm (NEW)
@@ -923,6 +923,136 @@ 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) =====
// ===== SESSION CRISIS TRACKING (#35) =====
var sessionCrisis = {
currentLevel: 0, // 0=NONE, 1=LOW, 2=MEDIUM, 3=HIGH, 4=CRITICAL
peakLevel: 0,
messageCount: 0,
history: [], // [{level, timestamp}]
escalationRate: 0, // levels per message
lastEscalation: null // timestamp of last escalation
};
function trackCrisis(text) {
var level = getCrisisLevel(text);
sessionCrisis.messageCount++;
sessionCrisis.history.push({ level: level, time: Date.now() });
if (level > sessionCrisis.currentLevel) {
sessionCrisis.lastEscalation = Date.now();
}
sessionCrisis.currentLevel = Math.max(sessionCrisis.currentLevel, level);
sessionCrisis.peakLevel = Math.max(sessionCrisis.peakLevel, level);
// Calculate escalation rate over last 5 messages
var recent = sessionCrisis.history.slice(-5);
if (recent.length >= 2) {
var first = recent[0].level;
var last = recent[recent.length - 1].level;
sessionCrisis.escalationRate = (last - first) / recent.length;
}
return getSessionContext();
}
function getSessionContext() {
var ctx = '';
if (sessionCrisis.history.length < 2) return ctx;
// Escalation detection: LOW -> HIGH in <= 3 messages
if (sessionCrisis.escalationRate > 0.5 && sessionCrisis.history.length <= 3) {
ctx += 'ESCALATION ALERT: User crisis level is rising rapidly. ';
}
// De-escalation detection: CRITICAL -> LOW for 5+ messages
if (sessionCrisis.peakLevel >= 3 && sessionCrisis.currentLevel <= 1 && sessionCrisis.messageCount >= 5) {
ctx += 'DE-ESCALATION: User appears to be calming. Maintain presence but reduce urgency. ';
}
// Persistent high state
if (sessionCrisis.currentLevel >= 2 && sessionCrisis.messageCount >= 3) {
ctx += 'User has been in crisis for ' + sessionCrisis.messageCount + ' messages. ';
}
// Level history summary
var levels = sessionCrisis.history.map(function(h) { return h.level; });
if (levels.length >= 2) {
ctx += 'Crisis trajectory: ' + levels.join(' → ') + '. ';
}
return ctx;
}
function resetSessionCrisis() {
sessionCrisis = {
currentLevel: 0, peakLevel: 0, messageCount: 0,
history: [], escalationRate: 0, lastEscalation: null
};
}
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';
var sessionCtx = getSessionContext();
var sessionBlock = sessionCtx ? '\nSESSION CONTEXT: ' + sessionCtx : '';
return SYSTEM_PROMPT + divider + header + directive + tone + sessionBlock;
}
// ===== OVERLAY =====
function showOverlay() {
crisisOverlay.classList.add('active');
@@ -1110,8 +1240,10 @@ Sovereignty and service always.`;
addMessage('user', text);
messages.push({ role: 'user', content: text });
var lastUserMessage = text;
checkCrisis(text);
trackCrisis(text);
msgInput.value = '';
msgInput.style.height = 'auto';
@@ -1126,7 +1258,7 @@ Sovereignty and service always.`;
sendBtn.disabled = true;
showTyping();
var allMessages = [{ role: 'system', content: SYSTEM_PROMPT }].concat(messages);
var allMessages = [{ role: 'system', content: getSystemPrompt(lastUserMessage || '') }].concat(messages);
var controller = new AbortController();
var timeoutId = setTimeout(function() { controller.abort(); }, 60000);

View File

@@ -1,5 +1,5 @@
[pytest]
testpaths = crisis
python_files = tests.py
python_files = tests.py tests_*.py
python_classes = Test*
python_functions = test_*

View File

@@ -0,0 +1,176 @@
"""
Test false positive fixes for issue #32.
Verifies that innocent messages no longer trigger crisis detection.
"""
import unittest
import sys
import os
# Ensure project root is on path
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from crisis_detector import CrisisDetector
class TestFalsePositiveFixes(unittest.TestCase):
"""Test that innocent messages no longer trigger false positives."""
def setUp(self):
self.detector = CrisisDetector()
def test_saying_goodbye_innocent(self):
"""'saying goodbye' should not trigger for innocent messages."""
innocent_messages = [
"I'm saying goodbye to my old apartment",
"She was saying goodbye at the airport",
"We were saying goodbye to our friends",
]
for msg in innocent_messages:
result = self.detector.scan(msg)
self.assertEqual(result.risk_level, "NONE",
f"False positive for: {msg}")
def test_wrote_a_will_innocent(self):
"""'wrote a will' should not trigger for responsible behavior."""
innocent_messages = [
"I wrote a will last year — just being responsible",
"She wrote a will as part of estate planning",
]
for msg in innocent_messages:
result = self.detector.scan(msg)
self.assertEqual(result.risk_level, "NONE",
f"False positive for: {msg}")
def test_giving_away_possessions_innocent(self):
"""'giving away possessions' should not trigger for decluttering."""
innocent_messages = [
"I'm giving away possessions — decluttering my house",
"She's giving away her old clothes",
]
for msg in innocent_messages:
result = self.detector.scan(msg)
self.assertEqual(result.risk_level, "NONE",
f"False positive for: {msg}")
def test_give_up_innocent(self):
"""'give up' should not trigger for common frustration."""
innocent_messages = [
"I give up on this project",
"Don't give up on your dreams",
"She gave up chocolate for Lent",
]
for msg in innocent_messages:
result = self.detector.scan(msg)
self.assertEqual(result.risk_level, "NONE",
f"False positive for: {msg}")
def test_trapped_innocent(self):
"""'trapped' should not trigger for literal situations."""
innocent_messages = [
"I'm trapped in an elevator",
"The cat is trapped in the tree",
"We're trapped in traffic",
]
for msg in innocent_messages:
result = self.detector.scan(msg)
self.assertEqual(result.risk_level, "NONE",
f"False positive for: {msg}")
def test_desperate_innocent(self):
"""'desperate' should not trigger for competition."""
innocent_messages = [
"She's desperate to win the game",
"He's desperate for a promotion",
]
for msg in innocent_messages:
result = self.detector.scan(msg)
self.assertEqual(result.risk_level, "NONE",
f"False positive for: {msg}")
def test_worthless_innocent(self):
"""'worthless' alone is MEDIUM but single-match falls to LOW — not HIGH."""
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}")
def test_hopeless_innocent(self):
"""'hopeless' alone is MEDIUM but single-match falls to LOW — not HIGH."""
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}")
def test_no_future_innocent(self):
"""'no future' should not trigger for political commentary."""
innocent_messages = [
"He has no future in politics",
"That company has no future",
]
for msg in innocent_messages:
result = self.detector.scan(msg)
self.assertEqual(result.risk_level, "NONE",
f"False positive for: {msg}")
def test_nothing_left_innocent(self):
"""'nothing left' should not trigger for everyday life."""
innocent_messages = [
"There's nothing left in the fridge",
"We have nothing left to do",
]
for msg in innocent_messages:
result = self.detector.scan(msg)
self.assertEqual(result.risk_level, "NONE",
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."""
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}")
def test_contextual_high_triggers(self):
"""Verify that contextual HIGH tier phrases still trigger."""
# These should trigger HIGH because they have crisis context
crisis_messages = [
("I feel so hopeless about my life", "HIGH"),
("I'm trapped in this situation and can't escape", "HIGH"),
("I feel desperate for help", "HIGH"),
("I want to give up on life", "HIGH"),
("There's no point in living anymore", "HIGH"),
]
for msg, expected_level in crisis_messages:
result = self.detector.scan(msg)
self.assertIn(result.risk_level, ("HIGH", "CRITICAL"),
f"Should trigger HIGH/CRITICAL for: {msg}, got {result.risk_level}")
def test_critical_still_triggers(self):
"""Verify that CRITICAL tier phrases still trigger."""
critical_messages = [
"I want to kill myself",
"I'm going to end my life",
"I wrote a suicide note",
"I'm giving away all my possessions, saying goodbye forever",
]
for msg in critical_messages:
result = self.detector.scan(msg)
self.assertEqual(result.risk_level, "CRITICAL",
f"Should trigger CRITICAL for: {msg}")
if __name__ == "__main__":
unittest.main()