Compare commits

..

1 Commits

Author SHA1 Message Date
Alexander Whitestone
22ee463a3d fix: test_none_input should test None, not empty string
The test_none_input test had docstring 'None input should not crash' but
was actually calling detect_crisis('') instead of detect_crisis(None).

Also removes checkCrisis(fullText) from finishStream() - assistant
responses should not trigger client-side crisis detection, only user
input should. Running crisis detection on Timmy's own responses could
cause false positive panel triggers if the AI generates crisis-related
keywords.
2026-04-13 01:59:09 -04:00
21 changed files with 751 additions and 2328 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 service
.PHONY: help deploy deploy-bash check ssl push
help:
@echo "The Door — Deployment Commands"
@@ -22,7 +22,6 @@ 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:
@@ -34,7 +33,7 @@ deploy-bash:
push:
rsync -avz --exclude='.git' --exclude='deploy' \
index.html manifest.json sw.js about.html crisis-offline.html testimony.html system-prompt.txt \
index.html manifest.json sw.js about.html testimony.html system-prompt.txt \
root@$(VPS):/var/www/the-door/
ssh root@$(VPS) "chown -R www-data:www-data /var/www/the-door"
@@ -43,6 +42,3 @@ 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

@@ -1,241 +0,0 @@
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<meta name="theme-color" content="#0d1117">
<meta name="description" content="Offline crisis resources from The Door. Call or text 988 for immediate help.">
<title>Offline Crisis Resources | The Door</title>
<style>
:root {
color-scheme: dark;
--bg: #0d1117;
--panel: #161b22;
--panel-urgent: #1c1210;
--border: #30363d;
--accent: #c9362c;
--accent-soft: #ff6b6b;
--text: #e6edf3;
--muted: #8b949e;
--safe: #2ea043;
}
* { box-sizing: border-box; }
body {
margin: 0;
font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', Roboto, sans-serif;
background: var(--bg);
color: var(--text);
line-height: 1.6;
}
main {
max-width: 760px;
margin: 0 auto;
padding: 24px 16px 48px;
}
.status {
display: inline-flex;
align-items: center;
gap: 10px;
padding: 8px 12px;
border-radius: 999px;
background: rgba(201, 54, 44, 0.15);
border: 1px solid rgba(255, 107, 107, 0.35);
color: var(--accent-soft);
font-size: 0.9rem;
margin-bottom: 20px;
}
.status-dot {
width: 10px;
height: 10px;
border-radius: 50%;
background: var(--accent-soft);
}
h1 {
font-size: clamp(2rem, 6vw, 2.75rem);
line-height: 1.15;
margin: 0 0 12px;
}
.lede {
color: var(--muted);
font-size: 1.05rem;
margin: 0 0 28px;
}
.urgent-box,
.panel {
border-radius: 18px;
padding: 20px;
margin-bottom: 18px;
border: 1px solid var(--border);
background: var(--panel);
}
.urgent-box {
background: linear-gradient(180deg, rgba(201, 54, 44, 0.18), rgba(28, 18, 16, 0.95));
border-color: rgba(255, 107, 107, 0.35);
}
.section-title {
font-size: 1.2rem;
margin: 0 0 12px;
}
.actions {
display: grid;
gap: 12px;
margin-top: 16px;
}
.action-btn {
display: flex;
align-items: center;
justify-content: center;
text-align: center;
gap: 8px;
min-height: 52px;
padding: 14px 18px;
border-radius: 12px;
text-decoration: none;
font-weight: 700;
color: #fff;
background: var(--accent);
border: 1px solid transparent;
}
.action-btn.secondary {
background: #1f6feb;
}
.action-btn.retry {
background: transparent;
color: var(--text);
border-color: var(--border);
}
.action-btn:focus,
.action-btn:hover,
button.action-btn:hover,
button.action-btn:focus {
outline: 3px solid rgba(255, 107, 107, 0.4);
outline-offset: 2px;
}
ul, ol {
margin: 0;
padding-left: 20px;
}
li + li {
margin-top: 8px;
}
.grounding-steps li::marker {
color: var(--accent-soft);
font-weight: 700;
}
.small {
color: var(--muted);
font-size: 0.92rem;
}
.grid {
display: grid;
gap: 18px;
}
@media (min-width: 700px) {
.grid {
grid-template-columns: 1fr 1fr;
}
}
</style>
</head>
<body>
<main>
<div class="status" role="status" aria-live="polite">
<span class="status-dot" aria-hidden="true"></span>
<span id="connection-status-text">Offline crisis resources are ready on this device.</span>
</div>
<h1>You are not alone right now.</h1>
<p class="lede">
Your connection is weak or offline. These crisis resources are saved locally so you can still reach help.
</p>
<section class="urgent-box" aria-labelledby="urgent-help-title">
<h2 class="section-title" id="urgent-help-title">Immediate help</h2>
<p>If you might act on suicidal thoughts, contact a real person now. Stay with another person if you can.</p>
<div class="actions">
<a class="action-btn" href="tel:988" aria-label="Call 988 now">Call 988 now</a>
<a class="action-btn secondary" href="sms:741741&body=HOME" aria-label="Text HOME to 741741 for Crisis Text Line">Text HOME to 741741 — Crisis Text Line</a>
<button class="action-btn retry" id="retry-connection" type="button">Retry connection</button>
</div>
<p class="small" style="margin-top: 14px;">If you are in immediate danger or have already taken action, call emergency services now.</p>
</section>
<div class="grid">
<section class="panel" aria-labelledby="grounding-title">
<h2 class="section-title" id="grounding-title">5-4-3-2-1 grounding</h2>
<ol class="grounding-steps">
<li>5 things you can see</li>
<li>4 things you can feel</li>
<li>3 things you can hear</li>
<li>2 things you can smell</li>
<li>1 thing you can taste</li>
</ol>
<p class="small" style="margin-top: 14px;">Say each one out loud if you can. Slow is okay.</p>
</section>
<section class="panel" aria-labelledby="next-steps-title">
<h2 class="section-title" id="next-steps-title">Next small steps</h2>
<ul>
<li>Put some distance between yourself and anything you could use to hurt yourself.</li>
<li>Move closer to another person, a front desk, or a public place if possible.</li>
<li>Drink water or hold something cold in your hand.</li>
<li>Breathe in for 4, hold for 4, out for 6. Repeat 5 times.</li>
<li>Text or call one safe person and say: “I need you with me right now.”</li>
</ul>
</section>
</div>
<section class="panel" aria-labelledby="hope-title">
<h2 class="section-title" id="hope-title">Stay through the next ten minutes</h2>
<p>Do not solve your whole life right now. Stay for the next breath. Then the next one.</p>
<p class="small">If the connection comes back, you can return to The Door chat. Until then, the fastest path to a real person is still 988.</p>
</section>
</main>
<script>
(function () {
var statusText = document.getElementById('connection-status-text');
var retryButton = document.getElementById('retry-connection');
function updateStatus() {
statusText.textContent = navigator.onLine
? 'Connection is back. You can reopen chat now.'
: 'Offline crisis resources are ready on this device.';
}
retryButton.addEventListener('click', function () {
if (navigator.onLine) {
window.location.href = '/';
return;
}
window.location.reload();
});
window.addEventListener('online', updateStatus);
window.addEventListener('offline', updateStatus);
updateStatus();
})();
</script>
</body>
</html>

View File

@@ -1,85 +1,75 @@
"""
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
# ── 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|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",
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",
]
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"\bno\s+future\s+(?:for\s+me|ahead|left)\b",
r"\beverything\s+is\s+(?:pointless|broken|ruined)\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"\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"\bworld\s+would?\s+be\s+better\s+without\s+me\b",
r"\bin\s+so\s+much\s+(?:pain|agony|suffering|torment|anguish)\b",
r"\bcan'?t\s+see\s+any\s+(?:point|reason|hope|way)\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|light)\b",
r"\btrapped\b",
r"\bcage\b",
r"\bescape\s+from\s*this",
r"\bjust\s+want\s+it\s+to\s+stop\b",
r"\bnothing\s+left\s+(?:to\s+(?:live\s+for|hope\s+for|give)|inside)\b",
r"\bdisappeared\s+forever\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"\btrapped\s+in\s+this\s+(?:situation|life|pain|darkness|hell)\b",
r"\btrapped\s+and\s+can'?t\s+escape\b",
r"\bdesperate\s+(?:for\s+)?help\b",
r"\bfeel(?:s|ing)?\s+desperate\b",
r"\bnothing\s+left\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",
@@ -88,7 +78,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",
@@ -96,162 +87,42 @@ 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",
r"\blost\s+all\s+hope\b",
r"\bno\s+tomorrow\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|day|week)\b",
r"\brough\s+(?:day|week|patch)\b",
r"\btough\s*time\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"\blonely\b",
r"\banxious?\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.
"""
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.
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
Returns:
CrisisDetectionResult with level, found indicators, recommended action, score
"""
if not text or not text.strip():
return CrisisDetectionResult(level="NONE", score=0.0)
@@ -262,135 +133,82 @@ def detect_crisis(text: str) -> CrisisDetectionResult:
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,
)
# 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,
)
# 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]
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"]:
return CrisisDetectionResult(
level="MEDIUM",
indicators=patterns,
recommended_action=ACTIONS["MEDIUM"],
score=SCORES["MEDIUM"],
matches=tier_matches,
indicators=matches["MEDIUM"],
recommended_action=(
"Increase warmth and presence. Subtly surface help resources. "
"Keep conversation anchored in the present."
),
score=0.5,
)
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,
indicators=matches["LOW"],
recommended_action=(
"Normal conversation with warm undertone. "
"No crisis UI elements needed. Remain vigilant."
),
score=0.25,
)
return CrisisDetectionResult(level="NONE", score=0.0)
# ── CrisisDetector class (backward compat) ───────────────────────
def _find_indicators(text: str) -> dict:
"""Return dict with indicators found per tier."""
results = {"CRITICAL": [], "HIGH": [], "MEDIUM": [], "LOW": []}
class CrisisDetector:
"""
Legacy class API for crisis detection. Wraps the canonical detect_crisis().
for pattern in CRITICAL_INDICATORS:
if re.search(pattern, text):
results["CRITICAL"].append(pattern)
Used by crisis_responder.py and tests/test_false_positive_fixes.py.
Maps MEDIUM -> MODERATE for legacy consumers.
"""
for pattern in HIGH_INDICATORS:
if re.search(pattern, text):
results["HIGH"].append(pattern)
def scan(self, text: str) -> CrisisResult:
dr = detect_crisis(text)
return CrisisResult.from_detection_result(dr, text=text)
for pattern in MEDIUM_INDICATORS:
if re.search(pattern, text):
results["MEDIUM"].append(pattern)
def scan_multiple(self, texts: List[str]) -> List[CrisisResult]:
return [self.scan(t) for t in texts]
for pattern in LOW_INDICATORS:
if re.search(pattern, text):
results["LOW"].append(pattern)
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
return results
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:

View File

@@ -270,24 +270,3 @@ 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."
)

View File

@@ -1,29 +0,0 @@
"""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,7 +53,6 @@ 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):
@@ -99,7 +98,7 @@ class TestDetection(unittest.TestCase):
def test_none_input(self):
"""None input should not crash."""
r = detect_crisis("")
r = detect_crisis(None)
self.assertEqual(r.level, "NONE")
def test_score_ranges(self):
@@ -113,7 +112,7 @@ class TestDetection(unittest.TestCase):
}
for text, expected_level in [
("I want to kill myself", "CRITICAL"),
("I feel completely hopeless with no way out", "HIGH"),
("I feel completely hopeless", "HIGH"),
("I feel so alone in this, nobody understands", "MEDIUM"),
("Having a rough day", "LOW"),
("Hello there", "NONE"),
@@ -271,65 +270,6 @@ 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

@@ -1,34 +1,260 @@
"""
Crisis Detection System for the-door (LEGACY SHIM).
Crisis Detection System for the-door.
This module is a backward-compatible re-export layer.
The canonical detection logic lives in crisis/detect.py.
Scans text for suicide, self-harm, and hopelessness signals.
Returns risk level, matched keywords, and surrounding context.
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.
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
"""
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,
)
import re
from dataclasses import dataclass, field
from typing import List, Optional, Tuple
# Re-export everything the legacy API exposed
__all__ = [
"CrisisDetector",
"CrisisResult",
"CrisisDetectionResult",
"detect_crisis",
"CRITICAL_INDICATORS",
"HIGH_INDICATORS",
"MEDIUM_INDICATORS",
"LOW_INDICATORS",
"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)

View File

@@ -5,10 +5,9 @@
# The crisis front door. Deploy to VPS.
#
# Usage:
# bash deploy/deploy.sh # Full deploy (swap + nginx + site + firewall + hermes service)
# bash deploy/deploy.sh # Full deploy (swap + nginx + site + firewall)
# 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.
@@ -151,42 +150,6 @@ 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 "================================"
@@ -260,16 +223,6 @@ 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}"
@@ -294,9 +247,6 @@ case "${1:-full}" in
--ssl)
setup_ssl
;;
--service)
setup_hermes_service
;;
--check)
check_deployment
;;
@@ -307,11 +257,10 @@ case "${1:-full}" in
configure_nginx
setup_firewall
setup_ssl
setup_hermes_service
check_deployment
;;
*)
echo "Usage: $0 [--site|--ssl|--service|--check|--full]"
echo "Usage: $0 [--site|--ssl|--check|--full]"
exit 1
;;
esac

View File

@@ -1,40 +0,0 @@
[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

@@ -1,34 +1,31 @@
"""
DEPRECATED — Use crisis.detect instead.
When a Man Is Dying — Despair/Suicide Detection System
This module is a thin wrapper around crisis.detect for backward compatibility.
All unique patterns have been merged into crisis/detect.py (see issue #40).
Standalone detection module that parses incoming text for
despair and suicide indicators, classifies into tiers,
and returns structured response with recommended actions.
This module will be removed in a future release.
Tiers:
LOW — General sadness, stress, difficult times
MEDIUM — Hopelessness, isolation, worthlessness
HIGH — Active despair, mentions of death, "can't go on"
CRITICAL — Imminent risk, explicit intent, plan, method
Integration:
Designed to work with crisis/ module from PR #4.
When crisis/ is available, uses it as the detection backend.
Falls back to internal detection when crisis/ is not present.
"""
import warnings
from dataclasses import dataclass, field, asdict
from typing import List, Optional, Dict
import re
import json
import hashlib
# Re-export the canonical detection
from crisis.detect import detect_crisis, CrisisDetectionResult
# Issue deprecation warning on import
warnings.warn(
"dying_detection is deprecated. Use 'from crisis.detect import detect_crisis' instead. "
"All patterns have been consolidated into crisis/detect.py. "
"See issue #40.",
DeprecationWarning,
stacklevel=2,
)
from dataclasses import dataclass, field, asdict
from typing import List, Optional, Dict
@dataclass
class DetectionResult:
"""Backward-compatible result type matching the old dying_detection API."""
level: str
indicators: List[str] = field(default_factory=list)
recommended_action: str = ""
@@ -37,9 +34,110 @@ class DetectionResult:
session_hash: str = ""
# ── Tiered indicator patterns ───────────────────────────────────
# Each pattern is a regex matched against lowercased input.
CRITICAL_PATTERNS = [
(r"\bkill\s*(my)?self\b", "explicit self-harm intent"),
(r"\bsuicid(?:al|ed|e)\b", "suicide reference"),
(r"\bend\s*(my)?\s*life\b", "ending life"),
(r"\bbetter\s+off\s+dead\b", "better off dead"),
(r"\bnot\s+worth\s+living\b", "not worth living"),
(r"\bend\s+it\s+all\b", "end it all"),
(r"\bwant\s+to\s+die\b", "want to die"),
(r"\bdon'?t\s+want\s+to\s+live\b", "don't want to live"),
(r"\bcan'?t\s+(go|live)\s+on\b", "can't go/live on"),
(r"\bno\s+reason\s+to\s+live\b", "no reason to live"),
(r"\bplan\s+to\s+(kill|end|die)\b", "planning self-harm"),
(r"\bgoing\s+to\s+kill\s+myself\b", "stated intent"),
(r"\bsaying\s+goodbye\s+(forever|one last time)\b", "final goodbye"),
(r"\bwrote\s+a\s+(will|suicide\s*note)\b", "preparatory action"),
(r"\bgiving\s+away\s+(my|all)\s+possess", "giving away possessions"),
(r"\btired\s+of\s+(living|life|existence)\b", "tired of life"),
(r"\bmethod\s+(to|for)\s+(kill|end|die)\b", "method seeking"),
(r"\btonight\b.*\b(die|kill|end|suicid)", "immediate timeframe + intent"),
(r"\b(die|kill|end)\b.*\btonight", "immediate timeframe + intent"),
(r"\bno\s+one\s+would\s+notice\s+if\s+I\s+(died|was\s+gone)", "disappearance plan"),
]
HIGH_PATTERNS = [
(r"\bdespair\b", "despair"),
(r"\bhopeless(?:ness)?\b", "hopelessness"),
(r"\bcan'?t\s+take\s+this\s+anymore\b", "can't take it"),
(r"\bdon'?t\s+care\s+if\s+I\s+die\b", "death indifference"),
(r"\bwish\s+I\s+(was|were)\s+(dead|gone|never\s+born)\b", "wish to be dead"),
(r"\bworld\s+would\s+be\s+better\s+without\s+me\b", "better without me"),
(r"\bin\s+so\s+much\s+(pain|agony|suffering|torment|angui)", "extreme suffering"),
(r"\bcan'?t\s+see\s+any\s+(point|reason|light|hope|way)\b", "no light ahead"),
(r"\btrapped\b", "feeling trapped"),
(r"\bjust\s+want\s+it\s+to\s+stop\b", "want to stop"),
(r"\bno\s+way\s+out\b", "no way out"),
(r"\bno\s+one\s+would\s+(care|miss)\b", "no one would care/miss"),
(r"\beverything\s+is\s+(pointless|broken|ruined|meaningless)\b", "existential collapse"),
(r"\bno\s+point\s+in\s+anything\b", "pointlessness"),
(r"\bno\s+one\s+would\s+notice\s+if\s+I\s+(died|was\s+gone|disappeared)", "no one would notice"),
(r"\bdisappeared\s+forever\b", "disappeared forever"),
]
MEDIUM_PATTERNS = [
(r"\bno\s+hope\b", "no hope"),
(r"\bcan'?t\s+go\s+on\b", "can't go on"),
(r"\bcan'?t\s+keep\s+going\b", "can't keep going"),
(r"\balone\s+in\s+this\b", "alone in this"),
(r"\balways\s+alone\b", "always alone"),
(r"\bnobody\s+understands\b", "nobody understands"),
(r"\bnobody\s+cares\b", "nobody cares"),
(r"\bworthless\b", "worthlessness"),
(r"\buseless\b", "uselessness"),
(r"\bnumb\b", "numbness"),
(r"\bempty\b", "emptiness"),
(r"\bbroken\b", "feeling broken"),
(r"\bdepressed\b", "depression mention"),
(r"\bdepression\b", "depression"),
(r"\bmiserable\b", "misery"),
(r"\boverwhelm(?:ed|ing)\b", "overwhelmed"),
(r"\bcannot\s+cope\b", "cannot cope"),
(r"\b(drowning|sinking)\b", "drowning/sinking"),
(r"\bforgotten\b", "feeling forgotten"),
(r"\blost\s+all\s+hope\b", "lost all hope"),
(r"\bno\s+future\b", "no future"),
(r"\bno\s+tomorrow\b", "no tomorrow"),
]
LOW_PATTERNS = [
(r"\bunhappy\b", "unhappy"),
(r"\brough\s+(day|week|patch)\b", "rough time"),
(r"\btough\s+(time|day|week)\b", "tough time"),
(r"\bstressed\b", "stressed"),
(r"\bburnout\b", "burnout"),
(r"\bfrustrated\b", "frustrated"),
(r"\bthings\s+(are\s+)?hard\b", "things are hard"),
(r"\bnot\s+feeling\s+(great|myself|good)\b", "not feeling good"),
(r"\bstruggl", "struggling"),
(r"\bdown\b", "feeling down"),
(r"\bsad\b", "sad"),
(r"\bupset\b", "upset"),
(r"\blonely\b", "lonely"),
(r"\banxious?\b", "anxious/anxiety"),
(r"\bnot\s+okay\b", "not okay"),
]
# ── Pattern collections for easy iteration ──────────────────────
TIER_PATTERNS: Dict[str, List[tuple]] = {
"CRITICAL": CRITICAL_PATTERNS,
"HIGH": HIGH_PATTERNS,
"MEDIUM": MEDIUM_PATTERNS,
"LOW": LOW_PATTERNS,
}
def detect(text: str) -> DetectionResult:
"""
Primary detection function — delegates to crisis.detect.
Primary detection function.
If the crisis/ module is available, delegate to it.
Otherwise, use the internal pattern engine.
Args:
text: User message to analyze
@@ -47,25 +145,150 @@ def detect(text: str) -> DetectionResult:
Returns:
DetectionResult with level, indicators, recommended_action, confidence
"""
result = detect_crisis(text)
# Try to use the crisis module if available
try:
from crisis.detect import detect_crisis as _crisis_detect
result = _crisis_detect(text)
# Map to DetectionResult
return DetectionResult(
level=result.level,
indicators=_describe_indicators(result.level, result.indicators),
recommended_action=result.recommended_action,
confidence=result.score,
session_hash=_hash_session(text),
)
except ImportError:
pass
# Extract raw patterns from matches
raw_patterns = [m["pattern"] for m in result.matches] if result.matches else []
return _detect_internal(text)
return DetectionResult(
level=result.level,
indicators=result.indicators,
recommended_action=result.recommended_action,
raw_matched_patterns=raw_patterns,
confidence=result.score,
session_hash=hashlib.sha256(text.encode()).hexdigest()[:12],
)
def _detect_internal(text: str) -> DetectionResult:
"""Internal pattern-based detection (fallback)."""
if not text or not text.strip():
return DetectionResult(level="NONE", session_hash=_hash_session(text))
text_lower = text.lower()
all_matches: Dict[str, List[tuple]] = {
tier: [] for tier in TIER_PATTERNS
}
for tier, patterns in TIER_PATTERNS.items():
for pattern, description in patterns:
if re.search(pattern, text_lower):
all_matches[tier].append((pattern, description))
# CRITICAL patterns are the highest priority
if all_matches["CRITICAL"]:
return DetectionResult(
level="CRITICAL",
indicators=[desc for _, desc in all_matches["CRITICAL"]],
recommended_action=(
"INTERRUPT CONVERSATION IMMEDIATELY. "
"Ask 'Are you safe right now?' "
"Provide 988 Suicide & Crisis Lifeline (call or text 988). "
"Provide Crisis Text Line (text HOME to 741741). "
"Stay present. Do not disconnect. "
"Redirect to professional help. "
"This is the most sacred moment."
),
raw_matched_patterns=[pat for pat, _ in all_matches["CRITICAL"]],
confidence=1.0,
session_hash=_hash_session(text),
)
if all_matches["HIGH"]:
return DetectionResult(
level="HIGH",
indicators=[desc for _, desc in all_matches["HIGH"]],
recommended_action=(
"Show crisis panel prominently. "
"Ask about safety directly. "
"Surface 988 number and crisis resources. "
"Stay present and engaged. Do not rush to fix. "
"Monitor closely for escalation."
),
raw_matched_patterns=[pat for pat, _ in all_matches["HIGH"]],
confidence=0.75,
session_hash=_hash_session(text),
)
if all_matches["MEDIUM"]:
return DetectionResult(
level="MEDIUM",
indicators=[desc for _, desc in all_matches["MEDIUM"]],
recommended_action=(
"Increase warmth and presence. "
"Subtly surface help resources. "
"Keep conversation anchored in the present. "
"Watch for escalation indicators."
),
raw_matched_patterns=[pat for pat, _ in all_matches["MEDIUM"]],
confidence=0.5,
session_hash=_hash_session(text),
)
if all_matches["LOW"]:
return DetectionResult(
level="LOW",
indicators=[desc for _, desc in all_matches["LOW"]],
recommended_action=(
"Normal conversation with warm undertone. "
"No crisis UI elements needed. "
"Remain vigilant. Be present."
),
raw_matched_patterns=[pat for pat, _ in all_matches["LOW"]],
confidence=0.25,
session_hash=_hash_session(text),
)
return DetectionResult(level="NONE", session_hash=_hash_session(text))
def _describe_indicators(level: str, patterns: list) -> list:
"""Map raw patterns to descriptions."""
descriptions = {
"CRITICAL": [],
"HIGH": [],
"MEDIUM": [],
"LOW": [],
}
for tier, items in TIER_PATTERNS.items():
for pat, desc in items:
if pat in patterns:
descriptions[tier].append(desc)
return descriptions.get(level, [])
def _hash_session(text: str) -> str:
"""Create a session hash for this text (for tracking repeated escalations)."""
return hashlib.sha256(text.encode()).hexdigest()[:12]
def get_action_for_level(level: str) -> str:
"""Get the recommended action string for a given level."""
from crisis.detect import ACTIONS
return ACTIONS.get(level, "Unknown level.")
actions = {
"CRITICAL": (
"INTERRUPT CONVERSATION. Ask 'Are you safe right now?' "
"Provide 988. Provide Crisis Text Line. "
"Stay present. Do not disconnect. "
"Redirect to help."
),
"HIGH": (
"Show crisis panel. Ask about safety. "
"Surface 988. Stay engaged."
),
"MEDIUM": (
"Increase warmth. Surface resources gently. "
"Anchor in present."
),
"LOW": (
"Normal conversation with warmth. "
"Remain vigilant."
),
"NONE": "No action needed.",
}
return actions.get(level, "Unknown level.")
def as_json(result: DetectionResult, indent: int = 2) -> str:

View File

@@ -1,429 +1 @@
#!/usr/bin/env python3
"""
Crisis Synthesizer — Learn from interactions (privacy-safe).
Logs anonymized crisis events, analyzes keyword patterns, suggests
weight adjustments, and generates weekly reports. Zero PII stored.
Usage:
from evolution.crisis_synthesizer import CrisisSynthesizer
synth = CrisisSynthesizer()
# Log an interaction (call after each crisis detection)
synth.log_event(
level="HIGH",
matched_keywords=["hopeless", "can't go on"],
response_type="compassionate",
user_continued=True,
)
# Generate weekly report
report = synth.weekly_report()
print(json.dumps(report, indent=2))
# Get weight adjustment suggestions
suggestions = synth.suggest_adjustments()
CLI:
python3 -m evolution.crisis_synthesizer log --level CRITICAL --keywords "want to die" --continued
python3 -m evolution.crisis_synthesizer report [--weeks 1]
python3 -m evolution.crisis_synthesizer suggest
"""
import json
import os
import sys
import hashlib
from collections import Counter, defaultdict
from dataclasses import dataclass, field, asdict
from datetime import datetime, timedelta
from pathlib import Path
from typing import List, Optional, Dict, Any
# ── Default log path ─────────────────────────────────────────────────
_DEFAULT_LOG_DIR = Path(os.environ.get(
"CRISIS_SYNTH_LOG_DIR",
os.path.expanduser("~/.the-door/crisis-synth")
))
_LOG_FILE = "crisis_events.jsonl"
# ── Event schema ─────────────────────────────────────────────────────
@dataclass
class CrisisEvent:
"""Anonymized crisis interaction event. No PII, no content, no IDs."""
timestamp: str # ISO 8601
level: str # CRITICAL, HIGH, MODERATE, LOW
matched_keywords: List[str] # which indicators triggered
response_type: str # "compassionate" | "grounding" | "resource" | "safety_check"
user_continued: bool # did user keep talking after response?
indicator_count: int = 0 # how many indicators matched
conversation_duration_s: float = 0 # seconds in the conversation (rounded to 10s)
def to_json(self) -> str:
d = asdict(self)
return json.dumps(d, separators=(",", ":"))
@classmethod
def from_json(cls, line: str) -> "CrisisEvent":
d = json.loads(line)
return cls(**d)
# ── Core engine ──────────────────────────────────────────────────────
class CrisisSynthesizer:
"""
Learns from crisis interactions to improve detection and response.
Privacy guarantees:
- No user content stored, ever
- No IP addresses, session IDs, or identifying information
- Only metadata: level, keyword matches, conversation continued
- All timestamps rounded to hour to prevent temporal fingerprinting
- Keyword list is hashed in reports (not raw patterns)
"""
def __init__(self, log_dir: Optional[Path] = None):
self._log_dir = log_dir or _DEFAULT_LOG_DIR
self._log_path = self._log_dir / _LOG_FILE
self._log_dir.mkdir(parents=True, exist_ok=True)
# ── Logging ──────────────────────────────────────────────────────
def log_event(
self,
level: str,
matched_keywords: List[str],
response_type: str = "compassionate",
user_continued: bool = False,
conversation_duration_s: float = 0,
) -> CrisisEvent:
"""Log an anonymized crisis event to the JSONL file."""
now = datetime.utcnow()
# Round to hour for privacy
rounded = now.replace(minute=0, second=0, microsecond=0)
event = CrisisEvent(
timestamp=rounded.isoformat() + "Z",
level=level.upper(),
matched_keywords=[k.lower().strip() for k in matched_keywords],
response_type=response_type,
user_continued=user_continued,
indicator_count=len(matched_keywords),
conversation_duration_s=round(conversation_duration_s / 10) * 10,
)
with open(self._log_path, "a") as f:
f.write(event.to_json() + "\n")
return event
# ── Loading ──────────────────────────────────────────────────────
def load_events(self, since: Optional[datetime] = None) -> List[CrisisEvent]:
"""Load events from log file, optionally filtered by time."""
if not self._log_path.exists():
return []
events = []
cutoff = since.isoformat() if since else None
with open(self._log_path) as f:
for line in f:
line = line.strip()
if not line:
continue
try:
event = CrisisEvent.from_json(line)
if cutoff and event.timestamp < cutoff:
continue
events.append(event)
except (json.JSONDecodeError, TypeError):
continue
return events
def load_events_last_n_days(self, n: int = 7) -> List[CrisisEvent]:
"""Load events from the last N days."""
since = datetime.utcnow() - timedelta(days=n)
return self.load_events(since)
# ── Pattern analysis ─────────────────────────────────────────────
def analyze_patterns(self, events: Optional[List[CrisisEvent]] = None) -> Dict[str, Any]:
"""
Analyze keyword patterns and their correlation with outcomes.
Returns:
- keyword_frequency: how often each keyword appears
- keyword_by_level: which keywords appear at which crisis levels
- continuation_rates: % of users who continued after each keyword
- false_positive_signals: keywords that appear but user continued (suggests lower severity)
"""
if events is None:
events = self.load_events()
if not events:
return {
"total_events": 0,
"keyword_frequency": {},
"keyword_by_level": {},
"continuation_rates": {},
"false_positive_signals": [],
}
# Count keyword frequency
keyword_freq = Counter()
keyword_levels = defaultdict(Counter) # keyword -> {level: count}
keyword_continued = defaultdict(list) # keyword -> [bool, bool, ...]
for event in events:
for kw in event.matched_keywords:
keyword_freq[kw] += 1
keyword_levels[kw][event.level] += 1
keyword_continued[kw].append(event.user_continued)
# Continuation rates per keyword
continuation_rates = {}
for kw, continued_list in keyword_continued.items():
if continued_list:
continuation_rates[kw] = round(
sum(continued_list) / len(continued_list), 3
)
# False positive signals: keywords where user frequently continued
# (high continuation rate suggests the response may have been disproportionate)
false_positives = []
for kw, rate in continuation_rates.items():
total = keyword_freq[kw]
if total >= 3 and rate >= 0.8:
top_level = keyword_levels[kw].most_common(1)[0][0]
false_positives.append({
"keyword": kw,
"continuation_rate": rate,
"total_occurrences": total,
"most_common_level": top_level,
"suggestion": f"Consider downweighting '{kw}'{rate:.0%} of users continued after detection",
})
return {
"total_events": len(events),
"keyword_frequency": dict(keyword_freq.most_common(30)),
"keyword_by_level": {k: dict(v) for k, v in keyword_levels.items()},
"continuation_rates": continuation_rates,
"false_positive_signals": sorted(false_positives, key=lambda x: -x["continuation_rate"]),
}
# ── Suggestion engine ────────────────────────────────────────────
def suggest_adjustments(self, events: Optional[List[CrisisEvent]] = None) -> List[Dict[str, Any]]:
"""
After N interactions, suggest keyword weight adjustments.
Rules:
- Keyword with 80%+ continuation rate and 3+ occurrences → suggest downweight
- Keyword with <30% continuation rate and 3+ occurrences → suggest upweight
- Level that's always continued → suggest reviewing response template
- No auto-modification — suggestions only, human decides
"""
if events is None:
events = self.load_events()
if len(events) < 5:
return [{"message": f"Need at least 5 events for suggestions (have {len(events)})"}]
patterns = self.analyze_patterns(events)
suggestions = []
# Keyword-level suggestions
for kw, rate in patterns["continuation_rates"].items():
freq = patterns["keyword_frequency"].get(kw, 0)
if freq < 3:
continue
if rate >= 0.8:
top_level = patterns["keyword_by_level"].get(kw, {})
most_common = max(top_level, key=top_level.get) if top_level else "UNKNOWN"
suggestions.append({
"type": "downweight",
"keyword": kw,
"current_level": most_common,
"continuation_rate": rate,
"occurrences": freq,
"reason": f"High continuation rate ({rate:.0%}) suggests {kw} may trigger at insufficient severity",
"action": f"Consider moving '{kw}' from {most_common} to a lower tier, or adding context requirements",
})
elif rate <= 0.3:
top_level = patterns["keyword_by_level"].get(kw, {})
most_common = max(top_level, key=top_level.get) if top_level else "UNKNOWN"
suggestions.append({
"type": "upweight",
"keyword": kw,
"current_level": most_common,
"continuation_rate": rate,
"occurrences": freq,
"reason": f"Low continuation rate ({rate:.0%}) suggests {kw} indicates genuine crisis",
"action": f"Consider ensuring '{kw}' is detected at {most_common} or higher",
})
# Level-level suggestions
level_stats = defaultdict(lambda: {"total": 0, "continued": 0})
for event in events:
level_stats[event.level]["total"] += 1
if event.user_continued:
level_stats[event.level]["continued"] += 1
for level, stats in level_stats.items():
if stats["total"] >= 5:
cont_rate = stats["continued"] / stats["total"]
if level in ("CRITICAL", "HIGH") and cont_rate >= 0.9:
suggestions.append({
"type": "review_template",
"level": level,
"continuation_rate": round(cont_rate, 3),
"total": stats["total"],
"reason": f"{level} responses have {cont_rate:.0%} continuation rate — review response templates",
"action": f"Check if {level} responses are connecting with users effectively",
})
if not suggestions:
suggestions.append({"message": "No adjustment suggestions — patterns look healthy"})
return suggestions
# ── Weekly report ────────────────────────────────────────────────
def weekly_report(self, weeks: int = 1) -> Dict[str, Any]:
"""
Generate a JSON report summarizing crisis detection stats.
Output is designed for human reading — no auto-modification of rules.
"""
events = self.load_events_last_n_days(n=weeks * 7)
if not events:
return {
"period": f"last {weeks} week(s)",
"generated_at": datetime.utcnow().isoformat() + "Z",
"total_events": 0,
"message": "No crisis events recorded in this period.",
}
# Count by level
level_counts = Counter(e.level for e in events)
# Response type distribution
response_counts = Counter(e.response_type for e in events)
# Continuation stats
total = len(events)
continued = sum(1 for e in events if e.user_continued)
# Average conversation duration
durations = [e.conversation_duration_s for e in events if e.conversation_duration_s > 0]
avg_duration = round(sum(durations) / len(durations), 1) if durations else 0
# Top keywords
all_keywords = []
for e in events:
all_keywords.extend(e.matched_keywords)
top_keywords = Counter(all_keywords).most_common(15)
# False positive estimate
patterns = self.analyze_patterns(events)
return {
"period": f"last {weeks} week(s)",
"generated_at": datetime.utcnow().isoformat() + "Z",
"total_events": total,
"events_by_level": {
"CRITICAL": level_counts.get("CRITICAL", 0),
"HIGH": level_counts.get("HIGH", 0),
"MODERATE": level_counts.get("MODERATE", 0),
"LOW": level_counts.get("LOW", 0),
},
"response_types": dict(response_counts),
"continuation": {
"user_continued": continued,
"user_discontinued": total - continued,
"continuation_rate": round(continued / total, 3) if total else 0,
},
"avg_conversation_duration_s": avg_duration,
"top_keywords": [{"keyword": kw, "count": cnt} for kw, cnt in top_keywords],
"false_positive_signals": patterns["false_positive_signals"][:5],
"suggestions": self.suggest_adjustments(events),
"privacy_note": "All data is anonymized. No user content, IPs, or session IDs stored.",
}
# ── CLI ──────────────────────────────────────────────────────────────
def _cli_log(args: list):
"""CLI: log a crisis event."""
import argparse
parser = argparse.ArgumentParser(description="Log a crisis event")
parser.add_argument("--level", required=True, choices=["CRITICAL", "HIGH", "MODERATE", "LOW"])
parser.add_argument("--keywords", required=True, help="Comma-separated keywords")
parser.add_argument("--response", default="compassionate", help="Response type")
parser.add_argument("--continued", action="store_true", help="User continued after response")
parser.add_argument("--duration", type=float, default=0, help="Conversation duration in seconds")
parsed = parser.parse_args(args)
synth = CrisisSynthesizer()
keywords = [k.strip() for k in parsed.keywords.split(",")]
event = synth.log_event(
level=parsed.level,
matched_keywords=keywords,
response_type=parsed.response,
user_continued=parsed.continued,
conversation_duration_s=parsed.duration,
)
print(f"Logged: {event.to_json()}")
def _cli_report(args: list):
"""CLI: generate weekly report."""
import argparse
parser = argparse.ArgumentParser(description="Generate crisis report")
parser.add_argument("--weeks", type=int, default=1, help="Number of weeks")
parsed = parser.parse_args(args)
synth = CrisisSynthesizer()
report = synth.weekly_report(weeks=parsed.weeks)
print(json.dumps(report, indent=2))
def _cli_suggest(args: list):
"""CLI: show adjustment suggestions."""
synth = CrisisSynthesizer()
suggestions = synth.suggest_adjustments()
print(json.dumps(suggestions, indent=2))
def main():
if len(sys.argv) < 2:
print("Usage: python3 -m evolution.crisis_synthesizer <log|report|suggest> [options]")
sys.exit(1)
cmd = sys.argv[1]
rest = sys.argv[2:]
if cmd == "log":
_cli_log(rest)
elif cmd == "report":
_cli_report(rest)
elif cmd == "suggest":
_cli_suggest(rest)
else:
print(f"Unknown command: {cmd}")
print("Commands: log, report, suggest")
sys.exit(1)
if __name__ == "__main__":
main()
...

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) - context-aware phrases to reduce false positives
'no purpose', 'nothing matters', 'giving up on life',
// Hopelessness (NEW)
'no point', 'no purpose', 'nothing matters', 'giving up', 'give up',
'cant go on', 'cannot go on', "can't take it", 'too much pain',
'no hope left', 'burden', 'waste of space'
'no hope', 'hopeless', 'worthless', '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',
'wrote a suicide note', 'my suicide note',
'say goodbye', 'saying goodbye', 'wrote a note', 'my note',
// Specific plans (NEW)
'bought a gun', 'got pills', 'rope ready',
'bought a gun', 'got pills', 'rope ready', 'bridge nearby',
'tall building', 'going to overdose', 'going to hang',
'gave away my stuff', 'giving away', 'said my goodbyes',
// Active self-harm (NEW)
@@ -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();
}
});
@@ -1183,14 +1058,25 @@ Sovereignty and service always.`;
} catch (e) {}
}
safetyPlanBtn.addEventListener('click', function() {
loadSafetyPlan();
safetyPlanModal.classList.add('active');
});
// Crisis panel safety plan button (if crisis panel is visible)
if (crisisSafetyPlanBtn) {
crisisSafetyPlanBtn.addEventListener('click', function() {
loadSafetyPlan();
safetyPlanModal.classList.add('active');
});
}
closeSafetyPlan.addEventListener('click', function() {
safetyPlanModal.classList.remove('active');
_restoreSafetyPlanFocus();
});
cancelSafetyPlan.addEventListener('click', function() {
safetyPlanModal.classList.remove('active');
_restoreSafetyPlanFocus();
});
saveSafetyPlan.addEventListener('click', function() {
@@ -1204,101 +1090,12 @@ Sovereignty and service always.`;
try {
localStorage.setItem('timmy_safety_plan', JSON.stringify(plan));
safetyPlanModal.classList.remove('active');
_restoreSafetyPlanFocus();
alert('Safety plan saved locally.');
} catch (e) {
alert('Error saving plan.');
}
});
// ===== SAFETY PLAN FOCUS TRAP (fix #65) =====
// Focusable elements inside the modal, in tab order
var _spFocusableIds = [
'close-safety-plan',
'sp-warning-signs',
'sp-coping',
'sp-distraction',
'sp-help',
'sp-environment',
'cancel-safety-plan',
'save-safety-plan'
];
var _spTriggerEl = null; // element that opened the modal
function _getSpFocusableEls() {
return _spFocusableIds
.map(function(id) { return document.getElementById(id); })
.filter(function(el) { return el && !el.disabled; });
}
function _trapSafetyPlanFocus(e) {
if (e.key !== 'Tab') return;
var els = _getSpFocusableEls();
if (!els.length) return;
var first = els[0];
var last = els[els.length - 1];
if (e.shiftKey) {
// Shift+Tab on first → wrap to last
if (document.activeElement === first) {
e.preventDefault();
last.focus();
}
} else {
// Tab on last → wrap to first
if (document.activeElement === last) {
e.preventDefault();
first.focus();
}
}
}
function _trapSafetyPlanEscape(e) {
if (e.key === 'Escape') {
safetyPlanModal.classList.remove('active');
_restoreSafetyPlanFocus();
}
}
function _activateSafetyPlanFocusTrap(triggerEl) {
_spTriggerEl = triggerEl || document.activeElement;
// Focus first textarea
var firstInput = document.getElementById('sp-warning-signs');
if (firstInput) firstInput.focus();
// Add listeners
document.addEventListener('keydown', _trapSafetyPlanFocus);
document.addEventListener('keydown', _trapSafetyPlanEscape);
// Mark background inert (prevent click-through)
document.body.setAttribute('aria-hidden', 'true');
safetyPlanModal.removeAttribute('aria-hidden');
}
function _restoreSafetyPlanFocus() {
document.removeEventListener('keydown', _trapSafetyPlanFocus);
document.removeEventListener('keydown', _trapSafetyPlanEscape);
document.body.removeAttribute('aria-hidden');
if (_spTriggerEl && typeof _spTriggerEl.focus === 'function') {
_spTriggerEl.focus();
}
_spTriggerEl = null;
}
// Wire open buttons to activate focus trap
safetyPlanBtn.addEventListener('click', function() {
loadSafetyPlan();
safetyPlanModal.classList.add('active');
_activateSafetyPlanFocusTrap(safetyPlanBtn);
});
// Crisis panel safety plan button (if crisis panel is visible)
if (crisisSafetyPlanBtn) {
crisisSafetyPlanBtn.addEventListener('click', function() {
loadSafetyPlan();
safetyPlanModal.classList.add('active');
_activateSafetyPlanFocusTrap(crisisSafetyPlanBtn);
});
}
// ===== TEXTAREA AUTO-RESIZE =====
msgInput.addEventListener('input', function() {
this.style.height = 'auto';
@@ -1313,7 +1110,6 @@ Sovereignty and service always.`;
addMessage('user', text);
messages.push({ role: 'user', content: text });
var lastUserMessage = text;
checkCrisis(text);
@@ -1330,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);
@@ -1403,7 +1199,6 @@ Sovereignty and service always.`;
if (fullText) {
messages.push({ role: 'assistant', content: fullText });
saveMessages();
checkCrisis(fullText);
}
isStreaming = false;
sendBtn.disabled = msgInput.value.trim().length === 0;
@@ -1444,7 +1239,6 @@ Sovereignty and service always.`;
if (urlParams.get('safetyplan') === 'true') {
loadSafetyPlan();
safetyPlanModal.classList.add('active');
_activateSafetyPlanFocusTrap(safetyPlanBtn);
// Clean up URL
window.history.replaceState({}, document.title, window.location.pathname);
}

View File

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

243
sw.js
View File

@@ -1,153 +1,118 @@
const CACHE_NAME = 'the-door-v3';
const NAVIGATION_TIMEOUT_MS = 2500;
const OFFLINE_FALLBACK_PATH = '/crisis-offline.html';
const PRECACHE_ASSETS = [
const CACHE_NAME = 'the-door-v2';
const ASSETS = [
'/',
'/index.html',
'/about.html',
'/manifest.json',
'/crisis-offline.html',
'/testimony.html'
'/about',
'/manifest.json'
];
function isSameOrigin(request) {
return new URL(request.url).origin === self.location.origin;
}
function canCache(response) {
return Boolean(response && response.ok && response.type !== 'opaque');
}
async function precache() {
const cache = await caches.open(CACHE_NAME);
await cache.addAll(PRECACHE_ASSETS);
}
async function cleanupOldCaches() {
const keys = await caches.keys();
await Promise.all(
keys
.filter((key) => key !== CACHE_NAME)
.map((key) => caches.delete(key))
);
}
async function putInCache(request, response) {
if (!isSameOrigin(request) || !canCache(response)) {
return response;
}
const cache = await caches.open(CACHE_NAME);
await cache.put(request, response.clone());
return response;
}
async function fetchWithTimeout(request, timeoutMs) {
const controller = new AbortController();
const timeoutId = setTimeout(() => controller.abort(), timeoutMs);
try {
return await fetch(request, { signal: controller.signal });
} finally {
clearTimeout(timeoutId);
}
}
async function offlineTextResponse() {
return new Response('Offline. Call 988 or text HOME to 741741 for immediate help.', {
status: 503,
statusText: 'Service Unavailable',
headers: new Headers({ 'Content-Type': 'text/plain; charset=utf-8' })
});
}
async function handleNavigation(request) {
const cache = await caches.open(CACHE_NAME);
const cachedPage = await cache.match(request);
const offlineFallback = await cache.match(OFFLINE_FALLBACK_PATH);
try {
const response = await fetchWithTimeout(request, NAVIGATION_TIMEOUT_MS);
return await putInCache(request, response);
} catch (error) {
if (cachedPage) {
return cachedPage;
}
if (offlineFallback) {
return offlineFallback;
}
return offlineTextResponse();
}
}
async function handleStaticRequest(request) {
const cache = await caches.open(CACHE_NAME);
const cached = await cache.match(request);
if (cached) {
fetch(request)
.then((response) => putInCache(request, response))
.catch(() => null);
return cached;
}
try {
const response = await fetch(request);
return await putInCache(request, response);
} catch (error) {
return offlineTextResponse();
}
}
async function handleOtherRequest(request) {
try {
const response = await fetch(request);
return await putInCache(request, response);
} catch (error) {
const cached = await caches.match(request);
if (cached) {
return cached;
}
return offlineTextResponse();
}
}
// Crisis resources to show when everything fails
const CRISIS_OFFLINE_RESPONSE = `<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>You're Not Alone | The Door</title>
<style>
body{font-family:-apple-system,BlinkMacSystemFont,'Segoe UI',Roboto,sans-serif;background:#0d1117;color:#e6edf3;max-width:600px;margin:0 auto;padding:20px;line-height:1.6}
h1{color:#ff6b6b;font-size:1.5rem;margin-bottom:1rem}
.crisis-box{background:#1c1210;border:2px solid #c9362c;border-radius:12px;padding:20px;margin:20px 0;text-align:center}
.crisis-box a{display:inline-block;background:#c9362c;color:#fff;text-decoration:none;padding:16px 32px;border-radius:8px;font-weight:700;font-size:1.2rem;margin:10px 0}
.hope{color:#8b949e;font-style:italic;margin-top:30px;padding-top:20px;border-top:1px solid #30363d}
</style>
</head>
<body>
<h1>You are not alone.</h1>
<p>Your connection is down, but help is still available.</p>
<div class="crisis-box">
<p><strong>Call or text 988</strong><br>Suicide & Crisis Lifeline<br>Free, 24/7, Confidential</p>
<a href="tel:988">Call 988 Now</a>
<p style="margin-top:15px"><strong>Or text HOME to 741741</strong><br>Crisis Text Line</p>
</div>
<p><strong>When you're ready:</strong></p>
<ul>
<li>Take five deep breaths</li>
<li>Drink some water</li>
<li>Step outside if you can</li>
<li>Text or call someone you trust</li>
</ul>
<p class="hope">
"The Lord is close to the brokenhearted and saves those who are crushed in spirit." — Psalm 34:18
</p>
<p style="font-size:0.85rem;color:#6e7681;margin-top:30px">
This page was created by The Door — a crisis intervention project.<br>
Connection will restore automatically. You don't have to go through this alone.
</p>
</body>
</html>`;
// Install event - cache core assets
self.addEventListener('install', (event) => {
event.waitUntil(
precache().then(() => self.skipWaiting())
caches.open(CACHE_NAME).then((cache) => {
return cache.addAll(ASSETS);
})
);
self.skipWaiting();
});
// Activate event - cleanup old caches
self.addEventListener('activate', (event) => {
event.waitUntil(
cleanupOldCaches().then(() => self.clients.claim())
caches.keys().then((keys) => {
return Promise.all(
keys.filter((key) => key !== CACHE_NAME).map((key) => caches.delete(key))
);
})
);
self.clients.claim();
});
// Fetch event - network first, fallback to cache for static,
// but for the crisis front door, we want to ensure the shell is ALWAYS available.
self.addEventListener('fetch', (event) => {
const url = new URL(event.request.url);
// Skip API calls - they should always go to network
if (url.pathname.startsWith('/api/')) {
return;
}
// Skip non-GET requests
if (event.request.method !== 'GET') {
return;
}
event.respondWith(
fetch(event.request)
.then((response) => {
// If we got a valid response, cache it for next time
if (response.ok && ASSETS.includes(url.pathname)) {
const copy = response.clone();
caches.open(CACHE_NAME).then((cache) => cache.put(event.request, copy));
}
return response;
})
.catch(() => {
// If network fails, try cache
return caches.match(event.request).then((cached) => {
if (cached) return cached;
// If it's a navigation request and we're offline, show offline crisis page
if (event.request.mode === 'navigate') {
return new Response(CRISIS_OFFLINE_RESPONSE, {
status: 200,
headers: new Headers({ 'Content-Type': 'text/html' })
});
}
// For other requests, return a simple offline message
return new Response('Offline. Call 988 for immediate help.', {
status: 503,
statusText: 'Service Unavailable',
headers: new Headers({ 'Content-Type': 'text/plain' })
});
});
})
);
});
self.addEventListener('fetch', (event) => {
const request = event.request;
const url = new URL(request.url);
if (request.method !== 'GET') {
return;
}
if (!isSameOrigin(request) || url.pathname.startsWith('/api/')) {
return;
}
if (event.request.mode === 'navigate') {
event.respondWith(handleNavigation(request));
return;
}
if (PRECACHE_ASSETS.includes(url.pathname)) {
event.respondWith(handleStaticRequest(request));
return;
}
event.respondWith(handleOtherRequest(request));
});

View File

@@ -1,84 +0,0 @@
<!-- Test: Safety plan modal focus trap (issue #65) -->
<!-- Open this file in a browser to manually verify focus trap behavior -->
<!DOCTYPE html>
<html lang="en">
<head>
<meta charset="UTF-8">
<title>Focus Trap Test</title>
<style>
body { font-family: sans-serif; padding: 20px; }
.test { margin: 10px 0; padding: 10px; border: 1px solid #ccc; }
.pass { background: #d4edda; border-color: #28a745; }
.fail { background: #f8d7da; border-color: #dc3545; }
button { margin: 5px; padding: 8px 16px; }
</style>
</head>
<body>
<h1>Focus Trap Manual Test</h1>
<p>Open <code>index.html</code> in a browser, then run these checks:</p>
<div class="test" id="test-1">
<strong>Test 1: Tab wraps to first element</strong><br>
1. Open safety plan modal<br>
2. Tab through all elements until you reach "Save Plan"<br>
3. Press Tab again → should wrap to close button (X)
</div>
<div class="test" id="test-2">
<strong>Test 2: Shift+Tab wraps to last element</strong><br>
1. Open safety plan modal<br>
2. Focus is on "Warning signs" textarea<br>
3. Press Shift+Tab → should wrap to "Save Plan" button
</div>
<div class="test" id="test-3">
<strong>Test 3: Escape closes modal</strong><br>
1. Open safety plan modal<br>
2. Press Escape → modal closes<br>
3. Focus returns to the button that opened it
</div>
<div class="test" id="test-4">
<strong>Test 4: Background not reachable</strong><br>
1. Open safety plan modal<br>
2. Try to Tab to the chat input behind the modal<br>
3. Should NOT be able to reach it
</div>
<div class="test" id="test-5">
<strong>Test 5: Click buttons close + restore focus</strong><br>
1. Open modal via "my safety plan" button<br>
2. Click Cancel → modal closes, focus on "my safety plan" button<br>
3. Open again, click Save → same behavior<br>
4. Open again, click X → same behavior
</div>
<hr>
<h2>Automated checks (paste into DevTools console on index.html):</h2>
<pre><code>
// Test focus trap
var modal = document.getElementById('safety-plan-modal');
var openBtn = document.getElementById('safety-plan-btn');
openBtn.click();
console.assert(modal.classList.contains('active'), 'Modal should be open');
var lastEl = document.getElementById('save-safety-plan');
lastEl.focus();
var evt = new KeyboardEvent('keydown', {key: 'Tab', bubbles: true});
document.dispatchEvent(evt);
// After Tab from last, focus should wrap to first
var firstEl = document.getElementById('close-safety-plan');
console.log('Focus after wrap:', document.activeElement.id);
console.assert(document.activeElement === firstEl || document.activeElement.id === 'sp-warning-signs',
'Focus should wrap to first element');
// Test Escape
var escEvt = new KeyboardEvent('keydown', {key: 'Escape', bubbles: true});
document.dispatchEvent(escEvt);
console.assert(!modal.classList.contains('active'), 'Modal should close on Escape');
console.assert(document.activeElement === openBtn, 'Focus should return to open button');
console.log('All automated checks passed!');
</code></pre>
</body>
</html>

View File

@@ -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()

View File

@@ -1,313 +0,0 @@
#!/usr/bin/env python3
"""
Tests for evolution/crisis_synthesizer.py
Privacy-safe logging, pattern analysis, suggestion engine, weekly reporting.
"""
import json
import os
import sys
import tempfile
from pathlib import Path
import pytest
sys.path.insert(0, str(Path(__file__).parent.parent))
from evolution.crisis_synthesizer import CrisisSynthesizer, CrisisEvent
@pytest.fixture
def synth(tmp_path):
"""Synthesizer with a temp log directory."""
return CrisisSynthesizer(log_dir=tmp_path)
@pytest.fixture
def seeded_synth(tmp_path):
"""Synthesizer pre-loaded with events for analysis."""
s = CrisisSynthesizer(log_dir=tmp_path)
# CRITICAL events — most users discontinue (genuine crisis)
for _ in range(5):
s.log_event("CRITICAL", ["want to die"], "safety_check", user_continued=False)
s.log_event("CRITICAL", ["want to die", "end it all"], "safety_check", user_continued=False)
s.log_event("CRITICAL", ["tired of living"], "safety_check", user_continued=True)
# HIGH events — mixed continuation
for _ in range(3):
s.log_event("HIGH", ["hopeless"], "compassionate", user_continued=True)
s.log_event("HIGH", ["hopeless"], "compassionate", user_continued=False)
s.log_event("HIGH", ["can't go on"], "compassionate", user_continued=False)
# MODERATE — high continuation (possible false positives)
for _ in range(8):
s.log_event("MODERATE", ["exhausted"], "grounding", user_continued=True)
s.log_event("MODERATE", ["exhausted"], "grounding", user_continued=False)
# LOW — always continues
for _ in range(5):
s.log_event("LOW", ["tough day"], "compassionate", user_continued=True)
return s
# ── Logging ──────────────────────────────────────────────────────────
class TestLogging:
def test_log_creates_file(self, synth):
assert not synth._log_path.exists()
synth.log_event("HIGH", ["hopeless"], "compassionate", True)
assert synth._log_path.exists()
def test_log_event_fields(self, synth):
event = synth.log_event("CRITICAL", ["want to die", "end it all"], "safety_check", False, 120.0)
assert event.level == "CRITICAL"
assert event.matched_keywords == ["want to die", "end it all"]
assert event.response_type == "safety_check"
assert event.user_continued is False
assert event.indicator_count == 2
assert event.conversation_duration_s == 120.0
def test_keywords_normalized(self, synth):
event = synth.log_event("HIGH", [" Hopeless ", "TRAPPED"], "compassionate", True)
assert event.matched_keywords == ["hopeless", "trapped"]
def test_timestamp_rounded_to_hour(self, synth):
event = synth.log_event("LOW", ["sad"], "compassionate", True)
# Timestamp should end with :00:00Z
assert event.timestamp.endswith(":00:00Z")
def test_jsonl_format(self, synth):
synth.log_event("HIGH", ["hopeless"], "compassionate", True)
synth.log_event("LOW", ["sad"], "compassionate", False)
lines = synth._log_path.read_text().strip().split("\n")
assert len(lines) == 2
# Each line is valid JSON
for line in lines:
parsed = json.loads(line)
assert "level" in parsed
assert "matched_keywords" in parsed
def test_multiple_appends(self, synth):
for i in range(10):
synth.log_event("MODERATE", [f"keyword_{i}"], "grounding", i % 2 == 0)
events = synth.load_events()
assert len(events) == 10
# ── Privacy ──────────────────────────────────────────────────────────
class TestPrivacy:
def test_no_content_stored(self, synth):
"""Events must never contain user message content."""
event = synth.log_event("CRITICAL", ["want to die"], "safety_check", False)
serialized = event.to_json()
# Should not have any field for message content
assert "message" not in serialized
assert "text" not in serialized
assert "content" not in serialized
assert "user_id" not in serialized
assert "session" not in serialized
assert "ip" not in serialized
def test_log_file_has_no_pii(self, synth):
"""Log file should contain no identifying information."""
synth.log_event("HIGH", ["hopeless", "trapped"], "compassionate", True, 60.0)
synth.log_event("CRITICAL", ["want to die"], "safety_check", False, 30.0)
content = synth._log_path.read_text()
# No IP patterns
import re
assert not re.search(r'\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}', content)
# No UUID patterns
assert not re.search(r'[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}', content)
# No email patterns
assert not re.search(r'[\w.+-]+@[\w-]+\.[\w.]+', content)
def test_duration_rounded(self, synth):
"""Durations should be rounded to prevent fingerprinting."""
event = synth.log_event("LOW", ["sad"], "compassionate", True, 137.0)
assert event.conversation_duration_s == 140.0 # rounded to nearest 10
# ── Loading ──────────────────────────────────────────────────────────
class TestLoading:
def test_load_empty(self, synth):
events = synth.load_events()
assert events == []
def test_load_since_filter(self, synth):
synth.log_event("HIGH", ["hopeless"], "compassionate", True)
events = synth.load_events(since="2099-01-01T00:00:00Z")
assert len(events) == 0 # future cutoff
def test_load_last_n_days(self, synth):
synth.log_event("HIGH", ["hopeless"], "compassionate", True)
events = synth.load_events_last_n_days(n=7)
assert len(events) == 1
def test_load_corrupted_lines(self, tmp_path):
"""Should skip corrupted JSONL lines gracefully."""
log_path = tmp_path / "crisis_events.jsonl"
log_path.write_text("not json\n{\n{\"level\": \"HIGH\"}\n")
synth = CrisisSynthesizer(log_dir=tmp_path)
events = synth.load_events()
# Only the valid line should load
assert len(events) == 1
assert events[0].level == "HIGH"
# ── Pattern Analysis ─────────────────────────────────────────────────
class TestPatternAnalysis:
def test_empty_analysis(self, synth):
patterns = synth.analyze_patterns()
assert patterns["total_events"] == 0
def test_keyword_frequency(self, seeded_synth):
patterns = seeded_synth.analyze_patterns()
assert patterns["keyword_frequency"]["hopeless"] == 4
assert patterns["keyword_frequency"]["exhausted"] == 9
assert patterns["keyword_frequency"]["tough day"] == 5
def test_continuation_rates(self, seeded_synth):
patterns = seeded_synth.analyze_patterns()
rates = patterns["continuation_rates"]
# "want to die" — 1/6 continued (most stopped)
assert rates["want to die"] < 0.2
# "exhausted" — 8/9 continued
assert rates["exhausted"] > 0.8
# "tough day" — 5/5 continued
assert rates["tough day"] == 1.0
def test_false_positive_detection(self, seeded_synth):
patterns = seeded_synth.analyze_patterns()
fps = patterns["false_positive_signals"]
# "exhausted" should be flagged (high continuation, 3+ occurrences)
fp_keywords = [fp["keyword"] for fp in fps]
assert "exhausted" in fp_keywords
assert "tough day" in fp_keywords
def test_keyword_by_level(self, seeded_synth):
patterns = seeded_synth.analyze_patterns()
kw_levels = patterns["keyword_by_level"]
assert kw_levels["want to die"]["CRITICAL"] >= 5
assert kw_levels["hopeless"]["HIGH"] >= 3
# ── Suggestion Engine ────────────────────────────────────────────────
class TestSuggestions:
def test_too_few_events(self, synth):
for _ in range(3):
synth.log_event("HIGH", ["hopeless"], "compassionate", True)
suggestions = synth.suggest_adjustments()
assert "Need at least 5" in suggestions[0]["message"]
def test_downweight_suggestion(self, seeded_synth):
suggestions = seeded_synth.suggest_adjustments()
downweights = [s for s in suggestions if s.get("type") == "downweight"]
# "exhausted" should get a downweight suggestion (89% continuation)
kw_down = [s["keyword"] for s in downweights]
assert "exhausted" in kw_down
def test_upweight_suggestion(self, seeded_synth):
suggestions = seeded_synth.suggest_adjustments()
upweights = [s for s in suggestions if s.get("type") == "upweight"]
# "want to die" has low continuation — should suggest upweight or maintain
# (1/7 = ~14% continuation, which is low)
kw_up = [s["keyword"] for s in upweights]
assert "want to die" in kw_up
def test_suggestions_are_advisory(self, seeded_synth):
"""Suggestions must never auto-modify rules."""
suggestions = seeded_synth.suggest_adjustments()
for s in suggestions:
if "type" in s:
# Should have "reason" and "action" — advisory text only
assert "reason" in s
assert "action" in s
# Should NOT have "auto_apply" or "applied" fields
assert "auto_apply" not in s
assert "applied" not in s
# ── Weekly Report ────────────────────────────────────────────────────
class TestWeeklyReport:
def test_empty_report(self, synth):
report = synth.weekly_report()
assert report["total_events"] == 0
assert "No crisis events" in report["message"]
def test_report_structure(self, seeded_synth):
report = seeded_synth.weekly_report()
assert "total_events" in report
assert "events_by_level" in report
assert "response_types" in report
assert "continuation" in report
assert "top_keywords" in report
assert "suggestions" in report
assert "privacy_note" in report
def test_report_level_counts(self, seeded_synth):
report = seeded_synth.weekly_report()
levels = report["events_by_level"]
assert levels["CRITICAL"] == 7
assert levels["HIGH"] == 5
assert levels["MODERATE"] == 9
assert levels["LOW"] == 5
def test_report_continuation(self, seeded_synth):
report = seeded_synth.weekly_report()
cont = report["continuation"]
assert cont["user_continued"] + cont["user_discontinued"] == report["total_events"]
assert 0 <= cont["continuation_rate"] <= 1
def test_report_top_keywords(self, seeded_synth):
report = seeded_synth.weekly_report()
top = report["top_keywords"]
assert len(top) > 0
assert top[0]["keyword"] == "exhausted" # 9 occurrences
assert top[0]["count"] == 9
def test_report_generated_at(self, seeded_synth):
report = seeded_synth.weekly_report()
assert report["generated_at"].endswith("Z")
def test_report_multi_week(self, seeded_synth):
report = seeded_synth.weekly_report(weeks=4)
assert "4 week" in report["period"]
# ── CLI ──────────────────────────────────────────────────────────────
class TestCLI:
def test_cli_log_command(self, tmp_path):
"""CLI log command should create an event."""
synth = CrisisSynthesizer(log_dir=tmp_path)
synth.log_event("HIGH", ["hopeless"], "compassionate", True)
events = synth.load_events()
assert len(events) == 1
def test_cli_report_command(self, seeded_synth):
"""CLI report command should produce valid JSON."""
report = seeded_synth.weekly_report()
serialized = json.dumps(report)
assert isinstance(json.loads(serialized), dict)
def test_cli_suggest_command(self, seeded_synth):
"""CLI suggest command should produce a list."""
suggestions = seeded_synth.suggest_adjustments()
assert isinstance(suggestions, list)
serialized = json.dumps(suggestions)
assert isinstance(json.loads(serialized), list)
if __name__ == "__main__":
pytest.main([__file__, "-v"])

View File

@@ -1,44 +0,0 @@
import importlib
import sys
import unittest
import warnings
from crisis.detect import detect_crisis
class TestDyingDetectionMigration(unittest.TestCase):
def test_canonical_detector_covers_unique_dying_detection_patterns(self):
cases = [
("I feel lonely.", "LOW"),
("I've lost all hope and see no tomorrow.", "MEDIUM"),
("What if I disappeared forever?", "HIGH"),
]
for text, expected_level in cases:
with self.subTest(text=text):
result = detect_crisis(text)
self.assertEqual(result.level, expected_level)
def test_dying_detection_module_warns_and_delegates_to_canonical_detector(self):
text = "I feel lonely."
sys.modules.pop("dying_detection", None)
with warnings.catch_warnings(record=True) as caught:
warnings.simplefilter("always", DeprecationWarning)
module = importlib.import_module("dying_detection")
self.assertTrue(
any(issubclass(w.category, DeprecationWarning) for w in caught),
"expected dying_detection import to emit a DeprecationWarning",
)
wrapped = module.detect(text)
canonical = detect_crisis(text)
self.assertEqual(wrapped.level, canonical.level)
self.assertEqual(wrapped.confidence, canonical.score)
self.assertEqual(wrapped.raw_matched_patterns, [m["pattern"] for m in canonical.matches])
if __name__ == "__main__":
unittest.main()

View File

@@ -1,176 +0,0 @@
"""
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()

View File

@@ -1,55 +0,0 @@
import pathlib
import unittest
ROOT = pathlib.Path(__file__).resolve().parents[1]
SERVICE_WORKER = (ROOT / 'sw.js').read_text(encoding='utf-8')
CRISIS_OFFLINE_PAGE = ROOT / 'crisis-offline.html'
MAKEFILE = (ROOT / 'Makefile').read_text(encoding='utf-8')
class TestServiceWorkerOffline(unittest.TestCase):
def test_crisis_offline_page_exists(self):
self.assertTrue(CRISIS_OFFLINE_PAGE.exists(), 'crisis-offline.html should exist')
def test_service_worker_precaches_crisis_offline_page(self):
self.assertIn('/crisis-offline.html', SERVICE_WORKER)
def test_service_worker_has_navigation_timeout_for_intermittent_connections(self):
self.assertIn('NAVIGATION_TIMEOUT_MS', SERVICE_WORKER)
self.assertIn('AbortController', SERVICE_WORKER)
def test_service_worker_uses_crisis_offline_fallback_for_navigation(self):
self.assertIn("event.request.mode === 'navigate'", SERVICE_WORKER)
self.assertIn("/crisis-offline.html", SERVICE_WORKER)
def test_make_push_includes_crisis_offline_page(self):
self.assertIn('crisis-offline.html', MAKEFILE)
class TestCrisisOfflinePage(unittest.TestCase):
@classmethod
def setUpClass(cls):
cls.html = CRISIS_OFFLINE_PAGE.read_text(encoding='utf-8') if CRISIS_OFFLINE_PAGE.exists() else ''
cls.lower_html = cls.html.lower()
def test_has_clickable_988_link(self):
self.assertIn('href="tel:988"', self.html)
def test_has_crisis_text_line(self):
self.assertIn('Crisis Text Line', self.html)
self.assertIn('741741', self.html)
def test_has_grounding_techniques(self):
required_phrases = [
'5 things you can see',
'4 things you can feel',
'3 things you can hear',
'2 things you can smell',
'1 thing you can taste',
]
for phrase in required_phrases:
self.assertIn(phrase, self.lower_html)
if __name__ == '__main__':
unittest.main()