fix: rate-limit crisis overlay display (#100)

crisis/__init__.py

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

crisis/gateway.py

@@ -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.

index.html

@@ -681,6 +681,7 @@ html, body {
   <!-- Footer -->
   <footer id="footer">
     <a href="/about.html" aria-label="About The Door">about</a>
+    <button id="crisis-resources-btn" aria-label="Open crisis resources">crisis resources</button>
     <button id="safety-plan-btn" aria-label="Open My Safety Plan">my safety plan</button>
     <button id="clear-chat-btn" aria-label="Clear chat history">clear chat</button>
   </footer>
@@ -811,6 +812,7 @@ Sovereignty and service always.`;
   var overlayCallLink = document.querySelector('.overlay-call');
   var statusDot = document.querySelector('.status-dot');
   var statusText = document.getElementById('status-text');
+  var crisisResourcesBtn = document.getElementById('crisis-resources-btn');
   
   // Safety Plan Elements
   var safetyPlanBtn = document.getElementById('safety-plan-btn');
@@ -826,6 +828,9 @@ Sovereignty and service always.`;
   var isStreaming = false;
   var overlayTimer = null;
   var crisisPanelShown = false;
+  var CRISIS_OVERLAY_COOLDOWN_MS = 10 * 60 * 1000;
+  var CRISIS_OVERLAY_LAST_SHOWN_KEY = 'timmy_crisis_overlay_last_shown_at';
+  var CRISIS_OVERLAY_EVENT_LOG_KEY = 'timmy_crisis_overlay_event_log';
 
   // ===== SERVICE WORKER =====
   if ('serviceWorker' in navigator) {
@@ -853,6 +858,43 @@ Sovereignty and service always.`;
   window.addEventListener('offline', updateOnlineStatus);
   updateOnlineStatus();
 
+  function getLastOverlayShownAt() {
+    try {
+      return parseInt(localStorage.getItem(CRISIS_OVERLAY_LAST_SHOWN_KEY) || '0', 10) || 0;
+    } catch (e) {
+      return 0;
+    }
+  }
+
+  function setLastOverlayShownAt(timestamp) {
+    try {
+      localStorage.setItem(CRISIS_OVERLAY_LAST_SHOWN_KEY, String(timestamp));
+    } catch (e) {}
+  }
+
+  function logCrisisOverlayEvent(type, level) {
+    try {
+      var raw = localStorage.getItem(CRISIS_OVERLAY_EVENT_LOG_KEY);
+      var events = raw ? JSON.parse(raw) : [];
+      if (!Array.isArray(events)) events = [];
+      events.push({ type: type, level: level, at: Date.now() });
+      if (events.length > 20) events = events.slice(events.length - 20);
+      localStorage.setItem(CRISIS_OVERLAY_EVENT_LOG_KEY, JSON.stringify(events));
+    } catch (e) {}
+  }
+
+  function openCrisisResources() {
+    crisisPanelShown = true;
+    crisisPanel.classList.add('visible');
+    if (typeof crisisPanel.scrollIntoView === 'function') {
+      crisisPanel.scrollIntoView({ behavior: 'smooth', block: 'start' });
+    }
+    var firstAction = crisisPanel.querySelector('.crisis-btn, a[href]');
+    if (firstAction && typeof firstAction.focus === 'function') {
+      firstAction.focus();
+    }
+  }
+
   // ===== CRISIS KEYWORDS =====
   // Tier 1: General crisis indicators - triggers enhanced 988 panel
   var crisisKeywords = [
@@ -1021,6 +1063,19 @@ Sovereignty and service always.`;
   var _preOverlayFocusElement = null;
 
   function showOverlay() {
+    return showOverlayWithRateLimit(false, 2);
+  }
+
+  function showOverlayWithRateLimit(forceOpen, level) {
+    var lastShownAt = getLastOverlayShownAt();
+    if (!forceOpen && Date.now() - lastShownAt < CRISIS_OVERLAY_COOLDOWN_MS) {
+      logCrisisOverlayEvent('suppressed', level || 2);
+      return false;
+    }
+
+    logCrisisOverlayEvent(forceOpen ? 'manual-open' : 'shown', level || 2);
+    setLastOverlayShownAt(Date.now());
+
     // Save current focus for restoration on dismiss
     _preOverlayFocusElement = document.activeElement;
 
@@ -1053,6 +1108,7 @@ Sovereignty and service always.`;
 
     // Focus the Call 988 link (always enabled) — disabled buttons cannot receive focus
     if (overlayCallLink) overlayCallLink.focus();
+    return true;
   }
 
   // Register focus trap on document (always listening, gated by class check)
@@ -1301,6 +1357,12 @@ Sovereignty and service always.`;
     });
   }
 
+  if (crisisResourcesBtn) {
+    crisisResourcesBtn.addEventListener('click', function() {
+      openCrisisResources();
+    });
+  }
+
   // ===== TEXTAREA AUTO-RESIZE =====
   msgInput.addEventListener('input', function() {
     this.style.height = 'auto';

tests/test_crisis_overlay_rate_limit.py

@@ -0,0 +1,53 @@
+import pathlib
+import re
+import unittest
+
+ROOT = pathlib.Path(__file__).resolve().parents[1]
+INDEX_HTML = ROOT / 'index.html'
+
+
+class TestCrisisOverlayRateLimit(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls):
+        cls.html = INDEX_HTML.read_text()
+
+    def test_overlay_has_ten_minute_cooldown_constant(self):
+        self.assertRegex(
+            self.html,
+            r"CRISIS_OVERLAY_COOLDOWN_MS\s*=\s*10\s*\*\s*60\s*\*\s*1000",
+            'Expected a 10-minute crisis overlay cooldown constant.',
+        )
+
+    def test_show_overlay_suppresses_repeat_with_logging(self):
+        self.assertRegex(
+            self.html,
+            r"function\s+logCrisisOverlayEvent\s*\(",
+            'Expected a crisis overlay event logger.',
+        )
+        self.assertRegex(
+            self.html,
+            r"if\s*\(!forceOpen\s*&&\s*Date\.now\(\)\s*-\s*lastShownAt\s*<\s*CRISIS_OVERLAY_COOLDOWN_MS\)",
+            'Expected showOverlay to suppress repeated auto-displays inside the cooldown window.',
+        )
+        self.assertRegex(
+            self.html,
+            r"logCrisisOverlayEvent\('suppressed'",
+            'Expected suppressed overlay attempts to be logged.',
+        )
+
+    def test_manual_crisis_resources_button_exists_and_bypasses_cooldown(self):
+        self.assertIn('id="crisis-resources-btn"', self.html)
+        self.assertRegex(
+            self.html,
+            r"function\s+openCrisisResources\s*\(",
+            'Expected a manual crisis resources opener.',
+        )
+        self.assertRegex(
+            self.html,
+            r"crisisResourcesBtn\.addEventListener\('click',\s*function\(\)\s*\{\s*openCrisisResources\(\);",
+            'Expected the footer button to wire into openCrisisResources().',
+        )
+
+
+if __name__ == '__main__':
+    unittest.main()

tests/test_unified_gateway.py

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

tests/test_unified_scorer.py

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

unified_scorer.py

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