Compare commits

..

5 Commits

Author SHA1 Message Date
Alexander Whitestone
2b91272f4e feat: CLI command to view crisis metrics summary (#136)
All checks were successful
Sanity Checks / sanity-test (pull_request) Successful in 9s
Smoke Test / smoke (pull_request) Successful in 14s
New crisis/metrics.py:
- CrisisEvent dataclass for individual events
- log_event(): append to ~/.the-door/crisis-metrics.jsonl
- load_events(): load last N days from JSONL
- compute_summary(): aggregate by level, top indicators,
  sessions affected, daily average, peak day
- MetricsSummary dataclass
- format_summary(): human-readable report with bar chart
- CLI: python3 -m crisis.metrics --summary/--json/--last 7d/--log

Makefile targets:
- make metrics    — weekly summary report
- make metrics-json — raw JSON export

Closes #136
2026-04-15 12:41:07 -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
15 changed files with 1025 additions and 1116 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"
@@ -46,3 +46,12 @@ ssl:
service:
ssh root@$(VPS) "cd /opt/the-door && bash deploy/deploy.sh --service"
# Crisis metrics
.PHONY: metrics metrics-json
metrics: ## Show crisis metrics summary (last 7 days)
python3 -m crisis.metrics --summary
metrics-json: ## Export crisis metrics as JSON
python3 -m crisis.metrics --json

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>

View File

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

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",

199
crisis/metrics.py Normal file
View File

@@ -0,0 +1,199 @@
"""Crisis metrics — aggregate detection data for operators.
Tracks crisis detection events and provides summary reports.
Usage:
python3 -m crisis.metrics --summary # weekly report
python3 -m crisis.metrics --json # raw JSON export
python3 -m crisis.metrics --last 7d # last 7 days
"""
from __future__ import annotations
import json
import os
import sys
import time
from collections import Counter
from dataclasses import dataclass, asdict
from pathlib import Path
from typing import Any, Dict, List, Optional
# Data directory for metrics storage
_DATA_DIR = Path(os.getenv("CRISIS_DATA_DIR", str(Path.home() / ".the-door")))
_METRICS_FILE = _DATA_DIR / "crisis-metrics.jsonl"
@dataclass
class CrisisEvent:
"""A single crisis detection event."""
timestamp: float
level: str # NONE, LOW, MODERATE, HIGH, CRITICAL
indicators: list
session_id: str = ""
source: str = "" # "chat", "gateway", "cli"
@dataclass
class MetricsSummary:
"""Aggregated metrics summary."""
period_days: int
total_events: int
by_level: Dict[str, int]
top_indicators: List[tuple]
sessions_affected: int
avg_daily: float
peak_day: str
peak_count: int
generated_at: str
def log_event(event: CrisisEvent) -> None:
"""Log a crisis event to the metrics file."""
_DATA_DIR.mkdir(parents=True, exist_ok=True)
with open(_METRICS_FILE, "a") as f:
f.write(json.dumps(asdict(event)) + "\n")
def load_events(days: int = 7) -> List[CrisisEvent]:
"""Load crisis events from the last N days."""
if not _METRICS_FILE.exists():
return []
cutoff = time.time() - (days * 86400)
events = []
try:
with open(_METRICS_FILE) as f:
for line in f:
line = line.strip()
if not line:
continue
data = json.loads(line)
if data.get("timestamp", 0) >= cutoff:
events.append(CrisisEvent(**data))
except (json.JSONDecodeError, KeyError):
pass
return events
def compute_summary(days: int = 7) -> MetricsSummary:
"""Compute metrics summary for the given period."""
events = load_events(days)
now = time.time()
# By level
by_level = Counter(e.level for e in events)
# Top indicators
indicator_counts = Counter()
for e in events:
for ind in e.indicators:
indicator_counts[ind] += 1
top_indicators = indicator_counts.most_common(10)
# Sessions
sessions = set(e.session_id for e in events if e.session_id)
# Peak day
from collections import defaultdict
daily = defaultdict(int)
for e in events:
day = time.strftime("%Y-%m-%d", time.localtime(e.timestamp))
daily[day] += 1
peak_day = max(daily, key=daily.get) if daily else "N/A"
peak_count = daily.get(peak_day, 0)
return MetricsSummary(
period_days=days,
total_events=len(events),
by_level=dict(by_level),
top_indicators=top_indicators,
sessions_affected=len(sessions),
avg_daily=round(len(events) / max(days, 1), 1),
peak_day=peak_day,
peak_count=peak_count,
generated_at=time.strftime("%Y-%m-%d %H:%M:%S"),
)
def format_summary(summary: MetricsSummary) -> str:
"""Format metrics summary as human-readable report."""
lines = [
"Crisis Metrics Summary",
"=" * 40,
f"Period: Last {summary.period_days} days",
f"Generated: {summary.generated_at}",
"",
f"Total events: {summary.total_events}",
f"Daily avg: {summary.avg_daily}",
f"Sessions: {summary.sessions_affected}",
f"Peak day: {summary.peak_day} ({summary.peak_count} events)",
"",
]
if summary.by_level:
lines.append("By severity:")
for level in ["CRITICAL", "HIGH", "MODERATE", "LOW", "NONE"]:
count = summary.by_level.get(level, 0)
if count > 0:
bar = "" * min(count, 30)
lines.append(f" {level:10s} {count:4d} {bar}")
lines.append("")
if summary.top_indicators:
lines.append("Top indicators:")
for indicator, count in summary.top_indicators[:5]:
lines.append(f" {indicator}: {count}")
lines.append("")
if summary.total_events == 0:
lines.append("No crisis events in this period.")
return "\n".join(lines)
def main():
import argparse
parser = argparse.ArgumentParser(description="Crisis metrics summary")
parser.add_argument("--summary", action="store_true", help="Print summary report")
parser.add_argument("--json", action="store_true", dest="as_json", help="Output JSON")
parser.add_argument("--last", default="7d", help="Time period (e.g., 7d, 30d)")
parser.add_argument("--log", nargs=2, metavar=("LEVEL", "INDICATOR"), help="Log a test event")
args = parser.parse_args()
# Parse period
period_str = args.last.rstrip("d")
try:
days = int(period_str)
except ValueError:
days = 7
# Log mode
if args.log:
level, indicator = args.log
event = CrisisEvent(
timestamp=time.time(),
level=level.upper(),
indicators=[indicator],
session_id="cli-test",
source="cli",
)
log_event(event)
print(f"Logged: {level.upper()} / {indicator}")
return 0
# Compute summary
summary = compute_summary(days)
if args.as_json:
print(json.dumps(asdict(summary), indent=2))
else:
print(format_summary(summary))
return 0
if __name__ == "__main__":
sys.exit(main())

View File

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

View File

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

View File

@@ -1,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

@@ -965,76 +965,7 @@ Sovereignty and service always.`;
}
// ===== GET SYSTEM PROMPT (wraps with crisis context) =====
// ===== SESSION CRISIS TRACKING (#35) =====
var sessionCrisis = {
currentLevel: 0, // 0=NONE, 1=LOW, 2=MEDIUM, 3=HIGH, 4=CRITICAL
peakLevel: 0,
messageCount: 0,
history: [], // [{level, timestamp}]
escalationRate: 0, // levels per message
lastEscalation: null // timestamp of last escalation
};
function trackCrisis(text) {
var level = getCrisisLevel(text);
sessionCrisis.messageCount++;
sessionCrisis.history.push({ level: level, time: Date.now() });
if (level > sessionCrisis.currentLevel) {
sessionCrisis.lastEscalation = Date.now();
}
sessionCrisis.currentLevel = Math.max(sessionCrisis.currentLevel, level);
sessionCrisis.peakLevel = Math.max(sessionCrisis.peakLevel, level);
// Calculate escalation rate over last 5 messages
var recent = sessionCrisis.history.slice(-5);
if (recent.length >= 2) {
var first = recent[0].level;
var last = recent[recent.length - 1].level;
sessionCrisis.escalationRate = (last - first) / recent.length;
}
return getSessionContext();
}
function getSessionContext() {
var ctx = '';
if (sessionCrisis.history.length < 2) return ctx;
// Escalation detection: LOW -> HIGH in <= 3 messages
if (sessionCrisis.escalationRate > 0.5 && sessionCrisis.history.length <= 3) {
ctx += 'ESCALATION ALERT: User crisis level is rising rapidly. ';
}
// De-escalation detection: CRITICAL -> LOW for 5+ messages
if (sessionCrisis.peakLevel >= 3 && sessionCrisis.currentLevel <= 1 && sessionCrisis.messageCount >= 5) {
ctx += 'DE-ESCALATION: User appears to be calming. Maintain presence but reduce urgency. ';
}
// Persistent high state
if (sessionCrisis.currentLevel >= 2 && sessionCrisis.messageCount >= 3) {
ctx += 'User has been in crisis for ' + sessionCrisis.messageCount + ' messages. ';
}
// Level history summary
var levels = sessionCrisis.history.map(function(h) { return h.level; });
if (levels.length >= 2) {
ctx += 'Crisis trajectory: ' + levels.join(' → ') + '. ';
}
return ctx;
}
function resetSessionCrisis() {
sessionCrisis = {
currentLevel: 0, peakLevel: 0, messageCount: 0,
history: [], escalationRate: 0, lastEscalation: null
};
}
function getSystemPrompt(userText) {
function getSystemPrompt(userText) {
var level = getCrisisLevel(userText);
if (level === 0) return SYSTEM_PROMPT;
@@ -1047,19 +978,65 @@ function getSystemPrompt(userText) {
var directive = 'DIRECTIVE: ' + profile.directive + '\n';
var tone = 'TONE: ' + profile.tone + '\n';
var sessionCtx = getSessionContext();
var sessionBlock = sessionCtx ? '\nSESSION CONTEXT: ' + sessionCtx : '';
return SYSTEM_PROMPT + divider + header + directive + tone + sessionBlock;
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--;
@@ -1076,6 +1053,9 @@ function getSystemPrompt(userText) {
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');
@@ -1083,7 +1063,22 @@ function getSystemPrompt(userText) {
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;
}
});
@@ -1188,25 +1183,14 @@ function getSystemPrompt(userText) {
} 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() {
@@ -1220,12 +1204,101 @@ function getSystemPrompt(userText) {
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';
@@ -1243,7 +1316,6 @@ function getSystemPrompt(userText) {
var lastUserMessage = text;
checkCrisis(text);
trackCrisis(text);
msgInput.value = '';
msgInput.style.height = 'auto';
@@ -1372,6 +1444,7 @@ function getSystemPrompt(userText) {
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 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>

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