Compare commits

..

1 Commits

Author SHA1 Message Date
Alexander Whitestone
544dbba8cc fix: complete crisis overlay keyboard navigation (#95)
All checks were successful
Sanity Checks / sanity-test (pull_request) Successful in 8s
Smoke Test / smoke (pull_request) Successful in 13s
2026-04-17 01:52:32 -04:00
7 changed files with 102 additions and 296 deletions

View File

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

View File

@@ -2,21 +2,18 @@
Crisis Gateway Module for the-door.
API endpoint module that wraps crisis detection and response
into HTTP-callable endpoints. Integrates detect.py, unified_scorer.py, and response.py.
into HTTP-callable endpoints. Integrates detect.py and response.py.
Usage:
from crisis.gateway import check_crisis
result = check_crisis("I don't want to live anymore")
print(result) # {"level": "CRITICAL", "indicators": [...], "response": {...}}
"""
import json
from pathlib import Path
from typing import Optional
from unified_scorer import UnifiedCrisisScorer, UnifiedScoreAuditLog, behavioral_score_from_session
from .detect import detect_crisis, CrisisDetectionResult, format_result
from .compassion_router import router
from .response import (
@@ -53,74 +50,6 @@ def check_crisis(text: str) -> dict:
}
def check_crisis_multimodal(
text: str,
*,
tracker: Optional[CrisisSessionTracker] = None,
voice_score: Optional[float] = None,
image_score: Optional[float] = None,
behavioral_score: Optional[float] = None,
audit_log_path: Optional[Path] = None,
weights: Optional[dict] = None,
) -> dict:
"""Combine text, voice, image, and behavioral signals into one crisis assessment."""
detection = detect_crisis(text)
session_state = tracker.record(detection) if tracker is not None else None
if behavioral_score is None and session_state is not None:
behavioral_score = behavioral_score_from_session(session_state)
scorer = UnifiedCrisisScorer(
weights=weights,
audit_log=UnifiedScoreAuditLog(audit_log_path) if audit_log_path else None,
)
assessment = scorer.score(
text_score=detection.score,
voice_score=voice_score,
image_score=image_score,
behavioral_score=behavioral_score,
source_text=text,
)
unified_detection = CrisisDetectionResult(
level=assessment.level.value,
indicators=detection.indicators,
recommended_action=detection.recommended_action,
score=assessment.combined_score,
matches=detection.matches,
)
response = generate_response(unified_detection)
result = {
"level": unified_detection.level,
"score": unified_detection.score,
"indicators": detection.indicators,
"recommended_action": unified_detection.recommended_action,
"timmy_message": response.timmy_message,
"ui": {
"show_crisis_panel": response.show_crisis_panel,
"show_overlay": response.show_overlay,
"provide_988": response.provide_988,
},
"escalate": response.escalate,
"unified": {
"level": assessment.level.value,
"combined_score": assessment.combined_score,
"weights": assessment.weights,
"modalities": assessment.modalities,
"present_modalities": assessment.present_modalities,
},
}
if session_state is not None:
result["session"] = {
"current_level": session_state.current_level,
"peak_level": session_state.peak_level,
"message_count": session_state.message_count,
"is_escalating": session_state.is_escalating,
"is_deescalating": session_state.is_deescalating,
}
return result
def get_system_prompt(base_prompt: str, text: str = "") -> str:
"""
Sovereign Heart System Prompt Override.

View File

@@ -987,20 +987,33 @@ Sovereignty and service always.`;
// Focus trap: cycle through focusable elements within the crisis overlay
function getOverlayFocusableElements() {
return crisisOverlay.querySelectorAll(
return Array.prototype.slice.call(crisisOverlay.querySelectorAll(
'a[href], button:not([disabled]), [tabindex]:not([tabindex="-1"])'
);
));
}
function trapFocusInOverlay(e) {
if (!crisisOverlay.classList.contains('active')) return;
if (e.key === 'Escape') {
e.preventDefault();
closeOverlay(msgInput);
return;
}
if (e.key !== 'Tab') return;
var focusable = getOverlayFocusableElements();
if (focusable.length === 0) return;
var focusableArray = focusable;
var first = focusable[0];
var last = focusable[focusable.length - 1];
var activeIndex = focusableArray.indexOf(document.activeElement);
if (activeIndex === -1) {
e.preventDefault();
e.shiftKey ? last.focus() : first.focus();
return;
}
if (e.shiftKey) {
// Shift+Tab: if on first, wrap to last
@@ -1020,6 +1033,33 @@ Sovereignty and service always.`;
// Store the element that had focus before the overlay opened
var _preOverlayFocusElement = null;
function closeOverlay(restoreFocusTarget) {
crisisOverlay.classList.remove('active');
if (overlayTimer) {
clearInterval(overlayTimer);
overlayTimer = null;
}
// Re-enable background interaction
var mainApp = document.getElementById('app');
if (mainApp) mainApp.removeAttribute('inert');
var chatSection = document.getElementById('chat-area');
if (chatSection) chatSection.removeAttribute('aria-hidden');
var footerEl = document.querySelector('footer');
if (footerEl) footerEl.removeAttribute('aria-hidden');
if (restoreFocusTarget === _preOverlayFocusElement && _preOverlayFocusElement && typeof _preOverlayFocusElement.focus === 'function') {
_preOverlayFocusElement.focus();
} else if (restoreFocusTarget && typeof restoreFocusTarget.focus === 'function') {
restoreFocusTarget.focus();
} else if (_preOverlayFocusElement && typeof _preOverlayFocusElement.focus === 'function') {
_preOverlayFocusElement.focus();
} else {
msgInput.focus();
}
_preOverlayFocusElement = null;
}
function showOverlay() {
// Save current focus for restoration on dismiss
_preOverlayFocusElement = document.activeElement;
@@ -1030,10 +1070,10 @@ Sovereignty and service always.`;
overlayDismissBtn.textContent = 'Continue to chat (' + countdown + 's)';
// Disable background interaction via inert attribute
var mainApp = document.querySelector('.app');
var mainApp = document.getElementById('app');
if (mainApp) mainApp.setAttribute('inert', '');
// Also hide from assistive tech
var chatSection = document.getElementById('chat');
var chatSection = document.getElementById('chat-area');
if (chatSection) chatSection.setAttribute('aria-hidden', 'true');
var footerEl = document.querySelector('footer');
if (footerEl) footerEl.setAttribute('aria-hidden', 'true');
@@ -1060,27 +1100,7 @@ Sovereignty and service always.`;
overlayDismissBtn.addEventListener('click', function() {
if (!overlayDismissBtn.disabled) {
crisisOverlay.classList.remove('active');
if (overlayTimer) {
clearInterval(overlayTimer);
overlayTimer = null;
}
// Re-enable background interaction
var mainApp = document.querySelector('.app');
if (mainApp) mainApp.removeAttribute('inert');
var chatSection = document.getElementById('chat');
if (chatSection) chatSection.removeAttribute('aria-hidden');
var footerEl = document.querySelector('footer');
if (footerEl) footerEl.removeAttribute('aria-hidden');
// Restore focus to the element that had it before the overlay opened
if (_preOverlayFocusElement && typeof _preOverlayFocusElement.focus === 'function') {
_preOverlayFocusElement.focus();
} else {
msgInput.focus();
}
_preOverlayFocusElement = null;
closeOverlay(_preOverlayFocusElement);
}
});

View File

@@ -0,0 +1,54 @@
import pathlib
from playwright.sync_api import sync_playwright
ROOT = pathlib.Path(__file__).resolve().parents[1]
INDEX_HTML = ROOT / 'index.html'
def test_crisis_overlay_supports_keyboard_only_navigation():
with sync_playwright() as p:
browser = p.chromium.launch(headless=True)
page = browser.new_page()
try:
page.goto(INDEX_HTML.as_uri())
page.evaluate(
"""
() => {
window.fetch = () => Promise.reject(new Error('test fetch blocked'));
window.alert = () => {};
}
"""
)
page.fill('#msg-input', "I'm going to kill myself tonight")
page.press('#msg-input', 'Enter')
page.wait_for_function("document.getElementById('crisis-overlay').classList.contains('active')")
assert page.evaluate("document.activeElement.classList.contains('overlay-call')") is True
page.evaluate(
"""
() => {
const btn = document.getElementById('overlay-dismiss-btn');
btn.disabled = false;
btn.textContent = 'Continue to chat';
}
"""
)
page.keyboard.press('Tab')
assert page.evaluate("document.activeElement.id") == 'overlay-dismiss-btn'
page.keyboard.press('Tab')
assert page.evaluate("document.activeElement.classList.contains('overlay-call')") is True
page.keyboard.press('Shift+Tab')
assert page.evaluate("document.activeElement.id") == 'overlay-dismiss-btn'
page.keyboard.press('Escape')
page.wait_for_function("!document.getElementById('crisis-overlay').classList.contains('active')")
assert page.evaluate("document.activeElement.id") == 'msg-input'
finally:
browser.close()

View File

@@ -1,19 +0,0 @@
from crisis.gateway import check_crisis_multimodal
from crisis.session_tracker import CrisisSessionTracker
def test_multimodal_gateway_uses_unified_score_for_988_ui(tmp_path):
tracker = CrisisSessionTracker()
result = check_crisis_multimodal(
"I want to kill myself tonight",
tracker=tracker,
voice_score=0.92,
image_score=0.6,
audit_log_path=tmp_path / "audit.jsonl",
)
assert result["unified"]["level"] == "CRITICAL"
assert result["ui"]["provide_988"] is True
assert result["ui"]["show_overlay"] is True
assert result["unified"]["modalities"]["voice"] == 0.92
assert result["unified"]["modalities"]["behavioral"] >= 0.0

View File

@@ -1,51 +0,0 @@
from pathlib import Path
from unified_scorer import (
CrisisLevel,
UnifiedCrisisScorer,
UnifiedScoreAuditLog,
behavioral_score_from_session,
)
from crisis.session_tracker import SessionState
def test_unified_scorer_renormalizes_available_modalities_and_escalates():
scorer = UnifiedCrisisScorer()
assessment = scorer.score(
text_score=1.0,
voice_score=0.8,
image_score=None,
behavioral_score=0.7,
)
assert assessment.level is CrisisLevel.CRITICAL
assert assessment.combined_score > 0.8
assert assessment.present_modalities == ["text", "voice", "behavioral"]
def test_behavioral_score_rises_for_escalating_session_state():
session = SessionState(
current_level="HIGH",
peak_level="CRITICAL",
message_count=4,
level_history=["LOW", "MEDIUM", "HIGH", "CRITICAL"],
is_escalating=True,
is_deescalating=False,
escalation_rate=1.0,
consecutive_low_messages=0,
)
assert behavioral_score_from_session(session) >= 0.8
def test_audit_log_persists_anonymized_score_entries(tmp_path):
log_path = tmp_path / "unified-score-audit.jsonl"
scorer = UnifiedCrisisScorer(audit_log=UnifiedScoreAuditLog(log_path))
scorer.score(text_score=0.75, voice_score=0.2, image_score=0.1, behavioral_score=0.6, source_text="I feel trapped and hopeless")
lines = log_path.read_text().strip().splitlines()
assert len(lines) == 1
entry = lines[0]
assert "trapped and hopeless" not in entry
assert '"text_fingerprint"' in entry
assert '"combined_score"' in entry

View File

@@ -1,126 +0,0 @@
"""Unified multimodal crisis scoring for the-door."""
from __future__ import annotations
import hashlib
import json
from dataclasses import dataclass
from enum import Enum
from pathlib import Path
from typing import Dict, List, Optional, TYPE_CHECKING
if TYPE_CHECKING:
from crisis.session_tracker import SessionState
SCORE_BY_LEVEL = {"NONE": 0.0, "LOW": 0.25, "MEDIUM": 0.5, "HIGH": 0.75, "CRITICAL": 1.0}
LEVEL_RANK = {"NONE": 0, "LOW": 1, "MEDIUM": 2, "HIGH": 3, "CRITICAL": 4}
class CrisisLevel(Enum):
NONE = "NONE"
LOW = "LOW"
MEDIUM = "MEDIUM"
HIGH = "HIGH"
CRITICAL = "CRITICAL"
DEFAULT_WEIGHTS: Dict[str, float] = {
"text": 0.4,
"voice": 0.25,
"behavioral": 0.2,
"image": 0.15,
}
@dataclass
class UnifiedAssessment:
level: CrisisLevel
combined_score: float
weights: Dict[str, float]
modalities: Dict[str, Optional[float]]
present_modalities: List[str]
class UnifiedScoreAuditLog:
def __init__(self, path: Path | str):
self.path = Path(path)
def record(self, assessment: UnifiedAssessment, source_text: str = "") -> None:
self.path.parent.mkdir(parents=True, exist_ok=True)
fingerprint = hashlib.sha256(source_text.encode("utf-8")).hexdigest()[:12] if source_text else None
payload = {
"level": assessment.level.value,
"combined_score": round(assessment.combined_score, 4),
"weights": assessment.weights,
"modalities": assessment.modalities,
"present_modalities": assessment.present_modalities,
"text_fingerprint": fingerprint,
}
with self.path.open("a", encoding="utf-8") as fh:
fh.write(json.dumps(payload, sort_keys=True) + "\n")
class UnifiedCrisisScorer:
def __init__(self, weights: Optional[Dict[str, float]] = None, audit_log: Optional[UnifiedScoreAuditLog] = None):
self.weights = dict(DEFAULT_WEIGHTS)
if weights:
self.weights.update(weights)
self.audit_log = audit_log
def _normalize(self, modalities: Dict[str, Optional[float]]) -> Dict[str, float]:
present = [name for name, score in modalities.items() if score is not None]
if not present:
return {}
total = sum(self.weights[name] for name in present)
return {name: self.weights[name] / total for name in present}
def _level_for_score(self, score: float) -> CrisisLevel:
if score > 0.8:
return CrisisLevel.CRITICAL
if score > 0.6:
return CrisisLevel.HIGH
if score > 0.4:
return CrisisLevel.MEDIUM
if score > 0.0:
return CrisisLevel.LOW
return CrisisLevel.NONE
def score(
self,
*,
text_score: Optional[float],
voice_score: Optional[float] = None,
image_score: Optional[float] = None,
behavioral_score: Optional[float] = None,
source_text: str = "",
) -> UnifiedAssessment:
modalities = {
"text": text_score,
"voice": voice_score,
"behavioral": behavioral_score,
"image": image_score,
}
normalized = self._normalize(modalities)
combined = 0.0
for name, weight in normalized.items():
combined += float(modalities[name]) * weight
assessment = UnifiedAssessment(
level=self._level_for_score(combined),
combined_score=combined,
weights=normalized,
modalities=modalities,
present_modalities=[name for name, score in modalities.items() if score is not None],
)
if self.audit_log:
self.audit_log.record(assessment, source_text=source_text)
return assessment
def behavioral_score_from_session(session: 'SessionState') -> float:
current = SCORE_BY_LEVEL.get(session.current_level, 0.0)
peak_bonus = 0.1 if LEVEL_RANK.get(session.peak_level, 0) >= LEVEL_RANK["HIGH"] else 0.0
escalation_bonus = 0.15 if session.is_escalating else 0.0
rate_bonus = min(max(session.escalation_rate, 0.0), 1.0) * 0.1
deescalation_penalty = 0.15 if session.is_deescalating else 0.0
return max(0.0, min(1.0, current + peak_bonus + escalation_bonus + rate_bonus - deescalation_penalty))