Compare commits

..

5 Commits

Author SHA1 Message Date
Timmy
722feae199 feat: Crisis detection A/B test framework
All checks were successful
Sanity Checks / sanity-test (pull_request) Successful in 10s
Smoke Test / smoke (pull_request) Successful in 14s
Closes #101

Adds A/B testing capability for crisis detection algorithms:

1. Feature flag: crisis detection algorithm A vs B
   - Variant A: Current canonical detector (crisis/detect.py)
   - Variant B: Enhanced detector (more sensitive to MEDIUM indicators)
   - Configurable traffic split (default 50/50)
   - Deterministic variant assignment via text hash

2. Logging: which variant triggered for each event
   - JSONL log file with event details
   - Privacy-preserving (text hashed, not stored)
   - Includes latency measurements

3. Metrics: false positive rate, detection latency per variant
   - Detection distribution by level
   - Average latency per variant
   - False positive rate (with human labeling)
   - Disagreement rate between variants

4. API:
   - CrisisABTester class for full control
   - detect_crisis_ab() convenience function
   - compare_results() for side-by-side analysis
   - label_event() for human review
   - get_report() for human-readable output

23 tests passing.
2026-04-15 10:49:51 -04:00
48f48c7f26 feat: cache offline crisis resources (refs #41) (#74)
All checks were successful
Smoke Test / smoke (push) Successful in 7s
Sanity Checks / sanity-test (pull_request) Successful in 17s
Smoke Test / smoke (pull_request) Successful in 19s
Merge PR #74 (squash)
2026-04-14 22:09:59 +00:00
da31288525 fix: deprecate dying_detection and consolidate crisis detection (#40) (#76)
All checks were successful
Smoke Test / smoke (push) Successful in 4s
Merge PR #76 (squash)
2026-04-14 22:08:29 +00:00
8efc858cd7 fix: add keyboard focus trap to crisis overlay (#80)
Some checks failed
Smoke Test / smoke (push) Has been cancelled
Merge PR #80 (squash)
2026-04-14 22:08:28 +00:00
611c1c8456 fix(a11y): Safety plan modal keyboard focus trap (#65) (#81)
Some checks failed
Smoke Test / smoke (push) Has been cancelled
Merge PR #81 (squash)
2026-04-14 22:08:24 +00:00
13 changed files with 1560 additions and 377 deletions

View File

@@ -34,7 +34,7 @@ deploy-bash:
push:
rsync -avz --exclude='.git' --exclude='deploy' \
index.html manifest.json sw.js about.html testimony.html system-prompt.txt \
index.html manifest.json sw.js about.html crisis-offline.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"

241
crisis-offline.html Normal file
View File

@@ -0,0 +1,241 @@
<!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>

387
crisis/ab_test.py Normal file
View File

@@ -0,0 +1,387 @@
"""
Crisis Detection A/B Testing Framework for the-door.
Provides feature-flagged A/B testing for crisis detection algorithms.
Variant A: Current canonical detector (crisis/detect.py)
Variant B: Enhanced detector with contextual scoring (configurable)
Logs which variant triggered for each event and tracks metrics:
- False positive rate per variant
- Detection latency per variant
- Detection distribution per variant
"""
import json
import time
import hashlib
import os
from dataclasses import dataclass, field, asdict
from typing import Dict, List, Optional, Callable, Any
from datetime import datetime, timezone
from enum import Enum
from crisis.detect import detect_crisis as detect_crisis_variant_a, CrisisDetectionResult
class Variant(Enum):
"""A/B test variants."""
A = "A" # Control: current canonical detector
B = "B" # Treatment: enhanced detector
@dataclass
class ABTestConfig:
"""Configuration for A/B test."""
enabled: bool = True
variant_b_percentage: float = 0.5 # 50% traffic to variant B
seed: Optional[str] = None # For deterministic assignment
log_file: str = "crisis_ab_test.jsonl"
metrics_file: str = "crisis_ab_metrics.json"
@dataclass
class DetectionEvent:
"""Single detection event for logging."""
event_id: str
timestamp: str
text_hash: str # Hash of input text (privacy-preserving)
variant: str
level: str
score: float
indicators: List[str]
latency_ms: float
metadata: Dict[str, Any] = field(default_factory=dict)
@dataclass
class VariantMetrics:
"""Metrics for a single variant."""
variant: str
total_events: int = 0
detections_by_level: Dict[str, int] = field(default_factory=lambda: {
"CRITICAL": 0, "HIGH": 0, "MEDIUM": 0, "LOW": 0, "NONE": 0
})
avg_latency_ms: float = 0.0
total_latency_ms: float = 0.0
false_positives: int = 0 # Requires human labeling
true_positives: int = 0 # Requires human labeling
false_positive_rate: Optional[float] = None # Calculated when labels available
@dataclass
class ABTestMetrics:
"""Aggregate metrics for A/B test."""
variant_a: VariantMetrics = field(default_factory=lambda: VariantMetrics("A"))
variant_b: VariantMetrics = field(default_factory=lambda: VariantMetrics("B"))
start_time: str = ""
end_time: str = ""
total_events: int = 0
disagreements: int = 0 # Cases where variants disagree
class CrisisABTester:
"""
A/B testing framework for crisis detection algorithms.
Usage:
tester = CrisisABTester()
result = tester.detect("I feel hopeless")
# Returns CrisisDetectionResult from assigned variant
# Logs event and updates metrics
"""
def __init__(self, config: Optional[ABTestConfig] = None):
self.config = config or ABTestConfig()
self.metrics = ABTestMetrics(start_time=datetime.now(timezone.utc).isoformat())
self._variant_b_detector: Optional[Callable] = None
self._event_log: List[DetectionEvent] = []
# Load existing metrics if file exists
if os.path.exists(self.config.metrics_file):
self._load_metrics()
def set_variant_b_detector(self, detector: Callable[[str], CrisisDetectionResult]):
"""Set the detector function for variant B."""
self._variant_b_detector = detector
def _assign_variant(self, text: str) -> Variant:
"""
Assign variant based on text hash (deterministic) or random.
Uses hash for consistent assignment of same/similar texts.
"""
if not self.config.enabled:
return Variant.A
# Use text hash for deterministic assignment
hash_input = text.strip().lower()
if self.config.seed:
hash_input = self.config.seed + hash_input
hash_val = int(hashlib.md5(hash_input.encode()).hexdigest(), 16)
normalized = (hash_val % 1000) / 1000.0 # 0.0 to 0.999
return Variant.B if normalized < self.config.variant_b_percentage else Variant.A
def _get_variant_b_result(self, text: str) -> CrisisDetectionResult:
"""
Get detection result from variant B.
Falls back to variant A if no variant B detector is set.
"""
if self._variant_b_detector:
return self._variant_b_detector(text)
# Default variant B: enhanced contextual scoring
# More sensitive to MEDIUM indicators (requires only 1 instead of 2)
from crisis.detect import _find_indicators, ACTIONS, SCORES
text_lower = text.lower()
matches = _find_indicators(text_lower)
if not matches:
return CrisisDetectionResult(level="NONE", score=0.0)
# CRITICAL and HIGH: same as variant A
for tier in ("CRITICAL", "HIGH"):
if matches[tier]:
tier_matches = matches[tier]
patterns = [m["pattern"] for m in tier_matches]
return CrisisDetectionResult(
level=tier,
indicators=patterns,
recommended_action=ACTIONS[tier],
score=SCORES[tier],
matches=tier_matches,
)
# MEDIUM: variant B requires only 1 indicator (vs 2 in variant A)
if matches["MEDIUM"]:
tier_matches = matches["MEDIUM"]
patterns = [m["pattern"] for m in tier_matches]
return CrisisDetectionResult(
level="MEDIUM",
indicators=patterns,
recommended_action=ACTIONS["MEDIUM"],
score=SCORES["MEDIUM"],
matches=tier_matches,
)
if matches["LOW"]:
tier_matches = matches["LOW"]
patterns = [m["pattern"] for m in tier_matches]
return CrisisDetectionResult(
level="LOW",
indicators=patterns,
recommended_action=ACTIONS["LOW"],
score=SCORES["LOW"],
matches=tier_matches,
)
return CrisisDetectionResult(level="NONE", score=0.0)
def detect(self, text: str, metadata: Optional[Dict] = None) -> CrisisDetectionResult:
"""
Run A/B test detection.
Args:
text: Input text to analyze
metadata: Optional metadata to attach to event log
Returns:
CrisisDetectionResult from assigned variant
"""
if not self.config.enabled:
return detect_crisis_variant_a(text)
variant = self._assign_variant(text)
start_time = time.perf_counter()
if variant == Variant.A:
result = detect_crisis_variant_a(text)
else:
result = self._get_variant_b_result(text)
latency_ms = (time.perf_counter() - start_time) * 1000
# Log event
event = DetectionEvent(
event_id=f"{int(time.time() * 1000)}-{hash(text) % 10000:04d}",
timestamp=datetime.now(timezone.utc).isoformat(),
text_hash=hashlib.sha256(text.encode()).hexdigest()[:16],
variant=variant.value,
level=result.level,
score=result.score,
indicators=result.indicators[:5], # Limit for privacy
latency_ms=round(latency_ms, 3),
metadata=metadata or {}
)
self._event_log.append(event)
self._log_event(event)
# Update metrics
self._update_metrics(variant, result, latency_ms)
return result
def _log_event(self, event: DetectionEvent):
"""Append event to JSONL log file."""
try:
with open(self.config.log_file, "a") as f:
f.write(json.dumps(asdict(event)) + "\n")
except Exception as e:
print(f"Warning: Could not log A/B test event: {e}")
def _update_metrics(self, variant: Variant, result: CrisisDetectionResult, latency_ms: float):
"""Update running metrics."""
vm = self.metrics.variant_a if variant == Variant.A else self.metrics.variant_b
vm.total_events += 1
vm.detections_by_level[result.level] = vm.detections_by_level.get(result.level, 0) + 1
vm.total_latency_ms += latency_ms
vm.avg_latency_ms = vm.total_latency_ms / vm.total_events
self.metrics.total_events += 1
def compare_results(self, text: str) -> Dict[str, CrisisDetectionResult]:
"""
Run both variants and return both results (for analysis).
Does not log to A/B test metrics.
"""
result_a = detect_crisis_variant_a(text)
result_b = self._get_variant_b_result(text)
return {"A": result_a, "B": result_b}
def get_disagreement_rate(self) -> float:
"""
Calculate disagreement rate from logged events.
Requires running detect() for same texts with both variants.
"""
if not self._event_log:
return 0.0
# Group by text_hash
by_text: Dict[str, Dict[str, str]] = {}
for event in self._event_log:
if event.text_hash not in by_text:
by_text[event.text_hash] = {}
by_text[event.text_hash][event.variant] = event.level
disagreements = sum(
1 for variants in by_text.values()
if "A" in variants and "B" in variants and variants["A"] != variants["B"]
)
return disagreements / len(by_text) if by_text else 0.0
def get_metrics(self) -> ABTestMetrics:
"""Get current metrics snapshot."""
self.metrics.end_time = datetime.now(timezone.utc).isoformat()
self.metrics.disagreements = int(self.get_disagreement_rate() * self.metrics.total_events)
return self.metrics
def save_metrics(self):
"""Save metrics to JSON file."""
try:
with open(self.config.metrics_file, "w") as f:
json.dump(asdict(self.get_metrics()), f, indent=2)
except Exception as e:
print(f"Warning: Could not save A/B test metrics: {e}")
def _load_metrics(self):
"""Load metrics from JSON file."""
try:
with open(self.config.metrics_file, "r") as f:
data = json.load(f)
# Reconstruct metrics from saved data
if "variant_a" in data:
self.metrics.variant_a = VariantMetrics(**data["variant_a"])
if "variant_b" in data:
self.metrics.variant_b = VariantMetrics(**data["variant_b"])
self.metrics.total_events = data.get("total_events", 0)
self.metrics.disagreements = data.get("disagreements", 0)
except Exception as e:
print(f"Warning: Could not load A/B test metrics: {e}")
def label_event(self, event_id: str, is_true_positive: bool):
"""
Label an event as true/false positive (requires human review).
Updates false positive rate metrics.
"""
for event in self._event_log:
if event.event_id == event_id:
vm = self.metrics.variant_a if event.variant == "A" else self.metrics.variant_b
if is_true_positive:
vm.true_positives += 1
else:
vm.false_positives += 1
# Recalculate false positive rate
total_labelled = vm.true_positives + vm.false_positives
if total_labelled > 0:
vm.false_positive_rate = vm.false_positives / total_labelled
self.save_metrics()
return
raise ValueError(f"Event {event_id} not found")
def get_report(self) -> str:
"""Generate human-readable A/B test report."""
m = self.get_metrics()
lines = [
"=" * 60,
"CRISIS DETECTION A/B TEST REPORT",
"=" * 60,
f"Period: {m.start_time} to {m.end_time}",
f"Total Events: {m.total_events}",
f"Disagreements: {m.disagreements}",
"",
"VARIANT A (Control - Current Detector):",
f" Events: {m.variant_a.total_events}",
f" Avg Latency: {m.variant_a.avg_latency_ms:.3f} ms",
f" Detection Distribution:",
]
for level, count in m.variant_a.detections_by_level.items():
pct = (count / m.variant_a.total_events * 100) if m.variant_a.total_events else 0
lines.append(f" {level}: {count} ({pct:.1f}%)")
if m.variant_a.false_positive_rate is not None:
lines.append(f" False Positive Rate: {m.variant_a.false_positive_rate:.1%}")
lines.extend([
"",
"VARIANT B (Treatment - Enhanced Detector):",
f" Events: {m.variant_b.total_events}",
f" Avg Latency: {m.variant_b.avg_latency_ms:.3f} ms",
f" Detection Distribution:",
])
for level, count in m.variant_b.detections_by_level.items():
pct = (count / m.variant_b.total_events * 100) if m.variant_b.total_events else 0
lines.append(f" {level}: {count} ({pct:.1f}%)")
if m.variant_b.false_positive_rate is not None:
lines.append(f" False Positive Rate: {m.variant_b.false_positive_rate:.1%}")
lines.append("=" * 60)
return "\n".join(lines)
# ── Module-level convenience ──────────────────────────────────────
_default_tester: Optional[CrisisABTester] = None
def get_ab_tester(config: Optional[ABTestConfig] = None) -> CrisisABTester:
"""Get or create the default A/B tester instance."""
global _default_tester
if _default_tester is None:
_default_tester = CrisisABTester(config)
return _default_tester
def detect_crisis_ab(text: str, metadata: Optional[Dict] = None) -> CrisisDetectionResult:
"""Convenience function for A/B tested crisis detection."""
return get_ab_tester().detect(text, metadata)

View File

@@ -51,13 +51,13 @@ HIGH_INDICATORS = [
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\b",
r"\bin\s+so\s+much\s+(?:pain|agony|suffering|torment)\b",
r"\bcan'?t\s+see\s+any\s+(?:point|reason)\b",
r"\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"\bescape\s+from\s*this",
r"\bjust\s+want\s+it\s+to\s+stop\b",
r"\btrapped\s+(?:in\s+(?:my|this|a\s+dark)|and\s+can'?t\s+escape)\b",
r"\bnothing\s+left\s+(?:to\s+(?:live\s+for|hope\s+for|give)|inside)\b",
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",
@@ -68,6 +68,8 @@ HIGH_INDICATORS = [
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",
]
@@ -99,6 +101,8 @@ MEDIUM_INDICATORS = [
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",
@@ -112,7 +116,7 @@ MEDIUM_INDICATORS = [
LOW_INDICATORS = [
r"\bunhappy\b",
r"\bdown\b",
r"\btough\s*time\b",
r"\btough\s*(?:time|day|week)\b",
r"\brough\s+(?:day|week|patch)\b",
r"\bstressed\b",
r"\bburnout\b",
@@ -122,6 +126,8 @@ LOW_INDICATORS = [
r"\btired\b",
r"\bsad\b",
r"\bupset\b",
r"\blonely\b",
r"\banxious?\b",
r"\bnot\s*(?:good|great|okay)\b",
r"\bthings\s*are\s*hard\b",
r"\bstruggling\b",

View File

@@ -1,31 +1,34 @@
"""
When a Man Is Dying — Despair/Suicide Detection System
DEPRECATED — Use crisis.detect instead.
Standalone detection module that parses incoming text for
despair and suicide indicators, classifies into tiers,
and returns structured response with recommended actions.
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).
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.
This module will be removed in a future release.
"""
import re
import json
import hashlib
import warnings
from dataclasses import dataclass, field, asdict
from typing import List, Optional, Dict
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,
)
@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 = ""
@@ -34,110 +37,9 @@ 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.
If the crisis/ module is available, delegate to it.
Otherwise, use the internal pattern engine.
Primary detection function — delegates to crisis.detect.
Args:
text: User message to analyze
@@ -145,150 +47,25 @@ def detect(text: str) -> DetectionResult:
Returns:
DetectionResult with level, indicators, recommended_action, confidence
"""
# 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
result = detect_crisis(text)
return _detect_internal(text)
# Extract raw patterns from matches
raw_patterns = [m["pattern"] for m in result.matches] if result.matches else []
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]
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 get_action_for_level(level: str) -> str:
"""Get the recommended action string for a given 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.")
from crisis.detect import ACTIONS
return ACTIONS.get(level, "Unknown level.")
def as_json(result: DetectionResult, indent: int = 2) -> str:

View File

@@ -808,7 +808,6 @@ Sovereignty and service always.`;
var crisisPanel = document.getElementById('crisis-panel');
var crisisOverlay = document.getElementById('crisis-overlay');
var overlayDismissBtn = document.getElementById('overlay-dismiss-btn');
var overlayCallLink = document.querySelector('#crisis-overlay .overlay-call');
var statusDot = document.querySelector('.status-dot');
var statusText = document.getElementById('status-text');
@@ -984,12 +983,60 @@ Sovereignty and service always.`;
// ===== 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--;
@@ -1003,13 +1050,12 @@ Sovereignty and service always.`;
}
}, 1000);
// Focus the Call 988 link — the first actionable (non-disabled) element.
// Disabled buttons are not valid focus targets (WCAG 2.4.3).
if (overlayCallLink) {
overlayCallLink.focus();
}
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');
@@ -1017,7 +1063,22 @@ Sovereignty and service always.`;
clearInterval(overlayTimer);
overlayTimer = null;
}
msgInput.focus();
// 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;
}
});
@@ -1122,25 +1183,14 @@ 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() {
@@ -1154,12 +1204,101 @@ 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';
@@ -1305,6 +1444,7 @@ 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
python_files = tests.py
testpaths = crisis tests
python_files = tests.py test_*.py
python_classes = Test*
python_functions = test_*

219
sw.js
View File

@@ -1,118 +1,153 @@
const CACHE_NAME = 'the-door-v2';
const ASSETS = [
const CACHE_NAME = 'the-door-v3';
const NAVIGATION_TIMEOUT_MS = 2500;
const OFFLINE_FALLBACK_PATH = '/crisis-offline.html';
const PRECACHE_ASSETS = [
'/',
'/index.html',
'/about',
'/manifest.json'
'/about.html',
'/manifest.json',
'/crisis-offline.html',
'/testimony.html'
];
// 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>`;
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();
}
}
// Install event - cache core assets
self.addEventListener('install', (event) => {
event.waitUntil(
caches.open(CACHE_NAME).then((cache) => {
return cache.addAll(ASSETS);
})
precache().then(() => self.skipWaiting())
);
self.skipWaiting();
});
// Activate event - cleanup old caches
self.addEventListener('activate', (event) => {
event.waitUntil(
caches.keys().then((keys) => {
return Promise.all(
keys.filter((key) => key !== CACHE_NAME).map((key) => caches.delete(key))
);
})
cleanupOldCaches().then(() => self.clients.claim())
);
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);
const request = event.request;
const url = new URL(request.url);
// Skip API calls - they should always go to network
if (url.pathname.startsWith('/api/')) {
if (request.method !== 'GET') {
return;
}
// Skip non-GET requests
if (event.request.method !== 'GET') {
if (!isSameOrigin(request) || url.pathname.startsWith('/api/')) {
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 (event.request.mode === 'navigate') {
event.respondWith(handleNavigation(request));
return;
}
// 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' })
});
}
if (PRECACHE_ASSETS.includes(url.pathname)) {
event.respondWith(handleStaticRequest(request));
return;
}
// 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' })
});
});
})
);
event.respondWith(handleOtherRequest(request));
});

View File

@@ -0,0 +1,84 @@
<!-- 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>

357
tests/test_crisis_ab.py Normal file
View File

@@ -0,0 +1,357 @@
"""
Tests for Crisis Detection A/B Testing Framework.
"""
import unittest
import os
import json
import tempfile
import shutil
from unittest.mock import patch, MagicMock
# Import from the crisis module
import sys
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from crisis.ab_test import (
CrisisABTester,
ABTestConfig,
Variant,
DetectionEvent,
VariantMetrics,
ABTestMetrics,
detect_crisis_ab,
)
from crisis.detect import CrisisDetectionResult
class TestABTestConfig(unittest.TestCase):
"""Test A/B test configuration."""
def test_default_config(self):
config = ABTestConfig()
self.assertTrue(config.enabled)
self.assertEqual(config.variant_b_percentage, 0.5)
self.assertIsNone(config.seed)
self.assertEqual(config.log_file, "crisis_ab_test.jsonl")
self.assertEqual(config.metrics_file, "crisis_ab_metrics.json")
def test_custom_config(self):
config = ABTestConfig(
enabled=False,
variant_b_percentage=0.3,
seed="test-seed",
log_file="custom.jsonl",
metrics_file="custom.json"
)
self.assertFalse(config.enabled)
self.assertEqual(config.variant_b_percentage, 0.3)
self.assertEqual(config.seed, "test-seed")
class TestVariantAssignment(unittest.TestCase):
"""Test variant assignment logic."""
def test_deterministic_assignment(self):
"""Same text should always get same variant with same seed."""
config = ABTestConfig(seed="test-seed")
tester = CrisisABTester(config)
text = "I feel hopeless today"
variant1 = tester._assign_variant(text)
variant2 = tester._assign_variant(text)
self.assertEqual(variant1, variant2)
def test_assignment_distribution(self):
"""With 50% split, roughly half should go to each variant."""
config = ABTestConfig(seed="test-seed")
tester = CrisisABTester(config)
variants_a = 0
variants_b = 0
test_texts = [f"test message {i}" for i in range(100)]
for text in test_texts:
variant = tester._assign_variant(text)
if variant == Variant.A:
variants_a += 1
else:
variants_b += 1
# Should be roughly 50/50 (allow some variance)
self.assertGreater(variants_a, 30)
self.assertGreater(variants_b, 30)
def test_disabled_returns_a(self):
"""When disabled, should always return variant A."""
config = ABTestConfig(enabled=False)
tester = CrisisABTester(config)
for i in range(10):
variant = tester._assign_variant(f"test {i}")
self.assertEqual(variant, Variant.A)
class TestDetection(unittest.TestCase):
"""Test A/B detection logic."""
def setUp(self):
self.temp_dir = tempfile.mkdtemp()
self.log_file = os.path.join(self.temp_dir, "test_ab.jsonl")
self.metrics_file = os.path.join(self.temp_dir, "test_metrics.json")
self.config = ABTestConfig(
seed="test-seed",
log_file=self.log_file,
metrics_file=self.metrics_file
)
self.tester = CrisisABTester(self.config)
def tearDown(self):
shutil.rmtree(self.temp_dir, ignore_errors=True)
def test_detect_returns_result(self):
"""detect() should return CrisisDetectionResult."""
result = self.tester.detect("I feel sad")
self.assertIsInstance(result, CrisisDetectionResult)
self.assertIn(result.level, ["NONE", "LOW", "MEDIUM", "HIGH", "CRITICAL"])
def test_detect_critical_text(self):
"""Critical text should be detected regardless of variant."""
result = self.tester.detect("I want to kill myself")
self.assertEqual(result.level, "CRITICAL")
def test_detect_none_text(self):
"""Non-crisis text should return NONE."""
result = self.tester.detect("The weather is nice today")
self.assertEqual(result.level, "NONE")
def test_disabled_uses_variant_a(self):
"""When disabled, should use variant A only."""
config = ABTestConfig(enabled=False, log_file=self.log_file, metrics_file=self.metrics_file)
tester = CrisisABTester(config)
result = tester.detect("I feel hopeless")
self.assertIsInstance(result, CrisisDetectionResult)
def test_logs_events(self):
"""detect() should log events to JSONL file."""
self.tester.detect("I feel sad")
self.tester.detect("I feel happy")
self.assertTrue(os.path.exists(self.log_file))
with open(self.log_file) as f:
lines = f.readlines()
self.assertEqual(len(lines), 2)
event = json.loads(lines[0])
self.assertIn("event_id", event)
self.assertIn("variant", event)
self.assertIn("level", event)
def test_updates_metrics(self):
"""detect() should update metrics."""
self.tester.detect("I feel hopeless") # Should trigger detection
self.tester.detect("Hello world") # Should not trigger
metrics = self.tester.get_metrics()
self.assertEqual(metrics.total_events, 2)
def test_variant_b_more_sensitive(self):
"""Variant B should be more sensitive to MEDIUM indicators."""
# Create tester that always assigns variant B
config = ABTestConfig(
variant_b_percentage=1.0, # 100% to B
seed="test-seed",
log_file=self.log_file,
metrics_file=self.metrics_file
)
tester_b = CrisisABTester(config)
# Single MEDIUM indicator - variant A would return LOW, variant B returns MEDIUM
result_b = tester_b.detect("I feel worthless")
# Compare with variant A
config_a = ABTestConfig(
variant_b_percentage=0.0, # 0% to B (100% to A)
seed="test-seed",
log_file=self.log_file + ".a",
metrics_file=self.metrics_file + ".a"
)
tester_a = CrisisABTester(config_a)
result_a = tester_a.detect("I feel worthless")
# Variant B should be at least as sensitive
level_order = {"NONE": 0, "LOW": 1, "MEDIUM": 2, "HIGH": 3, "CRITICAL": 4}
self.assertGreaterEqual(
level_order[result_b.level],
level_order[result_a.level]
)
class TestCompareResults(unittest.TestCase):
"""Test compare_results functionality."""
def test_compare_returns_both_variants(self):
tester = CrisisABTester()
results = tester.compare_results("I feel hopeless")
self.assertIn("A", results)
self.assertIn("B", results)
self.assertIsInstance(results["A"], CrisisDetectionResult)
self.assertIsInstance(results["B"], CrisisDetectionResult)
def test_compare_does_not_log(self):
"""compare_results should not log to A/B test metrics."""
temp_dir = tempfile.mkdtemp()
log_file = os.path.join(temp_dir, "test.jsonl")
metrics_file = os.path.join(temp_dir, "metrics.json")
config = ABTestConfig(log_file=log_file, metrics_file=metrics_file)
tester = CrisisABTester(config)
tester.compare_results("I feel sad")
# Log file should not exist (no events logged)
self.assertFalse(os.path.exists(log_file))
shutil.rmtree(temp_dir)
class TestMetrics(unittest.TestCase):
"""Test metrics tracking."""
def setUp(self):
self.temp_dir = tempfile.mkdtemp()
self.log_file = os.path.join(self.temp_dir, "test.jsonl")
self.metrics_file = os.path.join(self.temp_dir, "metrics.json")
self.config = ABTestConfig(
seed="test-seed",
log_file=self.log_file,
metrics_file=self.metrics_file
)
def tearDown(self):
shutil.rmtree(self.temp_dir, ignore_errors=True)
def test_metrics_track_events(self):
tester = CrisisABTester(self.config)
for i in range(10):
tester.detect(f"test message {i}")
metrics = tester.get_metrics()
self.assertEqual(metrics.total_events, 10)
def test_metrics_track_levels(self):
tester = CrisisABTester(self.config)
tester.detect("I want to kill myself") # CRITICAL
tester.detect("I feel hopeless") # HIGH or MEDIUM
tester.detect("Hello world") # NONE
metrics = tester.get_metrics()
total_detections = sum(metrics.variant_a.detections_by_level.values())
total_detections += sum(metrics.variant_b.detections_by_level.values())
self.assertEqual(total_detections, 3)
def test_save_and_load_metrics(self):
tester = CrisisABTester(self.config)
for i in range(5):
tester.detect(f"test {i}")
tester.save_metrics()
self.assertTrue(os.path.exists(self.metrics_file))
# Create new tester that loads saved metrics
tester2 = CrisisABTester(self.config)
self.assertEqual(tester2.metrics.total_events, 5)
class TestEventLabeling(unittest.TestCase):
"""Test event labeling for false positive tracking."""
def setUp(self):
self.temp_dir = tempfile.mkdtemp()
self.log_file = os.path.join(self.temp_dir, "test.jsonl")
self.metrics_file = os.path.join(self.temp_dir, "metrics.json")
self.config = ABTestConfig(
seed="test-seed",
log_file=self.log_file,
metrics_file=self.metrics_file
)
self.tester = CrisisABTester(self.config)
def tearDown(self):
shutil.rmtree(self.temp_dir, ignore_errors=True)
def test_label_event_updates_metrics(self):
self.tester.detect("I feel hopeless")
event_id = self.tester._event_log[0].event_id
self.tester.label_event(event_id, is_true_positive=False)
metrics = self.tester.get_metrics()
# Find which variant was assigned
variant = self.tester._event_log[0].variant
vm = metrics.variant_a if variant == "A" else metrics.variant_b
self.assertEqual(vm.false_positives, 1)
self.assertEqual(vm.false_positive_rate, 1.0)
def test_label_nonexistent_event_raises(self):
with self.assertRaises(ValueError):
self.tester.label_event("nonexistent-id", is_true_positive=True)
class TestReport(unittest.TestCase):
"""Test report generation."""
def test_report_format(self):
tester = CrisisABTester()
for i in range(5):
tester.detect(f"test message {i}")
report = tester.get_report()
self.assertIn("CRISIS DETECTION A/B TEST REPORT", report)
self.assertIn("VARIANT A", report)
self.assertIn("VARIANT B", report)
self.assertIn("Total Events: 5", report)
class TestConvenienceFunction(unittest.TestCase):
"""Test module-level convenience function."""
def test_detect_crisis_ab(self):
result = detect_crisis_ab("I feel sad")
self.assertIsInstance(result, CrisisDetectionResult)
def test_detect_crisis_ab_with_metadata(self):
result = detect_crisis_ab("I feel sad", metadata={"source": "test"})
self.assertIsInstance(result, CrisisDetectionResult)
class TestCustomVariantBDetector(unittest.TestCase):
"""Test custom variant B detector."""
def test_custom_detector(self):
"""Should use custom detector when set."""
def custom_detector(text: str) -> CrisisDetectionResult:
return CrisisDetectionResult(
level="HIGH",
indicators=["custom"],
score=0.9
)
tester = CrisisABTester(ABTestConfig(variant_b_percentage=1.0))
tester.set_variant_b_detector(custom_detector)
result = tester.detect("Hello world")
self.assertEqual(result.level, "HIGH")
self.assertEqual(result.indicators, ["custom"])
if __name__ == "__main__":
unittest.main()

View File

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

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

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