Compare commits
1 Commits
fix/101-cr
...
door/issue
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
1f744e846d |
@@ -1,152 +0,0 @@
|
||||
"""
|
||||
A/B Test Framework for Crisis Detection in the-door.
|
||||
|
||||
Allows running two crisis detection variants side-by-side with
|
||||
logged outcomes for comparison. No PII stored — only variant labels,
|
||||
levels, and timing.
|
||||
|
||||
Usage:
|
||||
from crisis.ab_testing import ABTestCrisisDetector
|
||||
|
||||
detector = ABTestCrisisDetector(variant_a=detect_v1, variant_b=detect_v2)
|
||||
result, variant = detector.detect("I feel hopeless")
|
||||
# result: CrisisDetectionResult
|
||||
# variant: "A" or "B"
|
||||
|
||||
# Get comparison metrics
|
||||
stats = detector.get_stats()
|
||||
# {"A": {"count": 100, "avg_latency_ms": 2.3, ...}, "B": {...}}
|
||||
"""
|
||||
|
||||
import os
|
||||
import random
|
||||
import time
|
||||
from dataclasses import dataclass, field
|
||||
from typing import Callable, Dict, List, Optional, Tuple
|
||||
|
||||
from .detect import CrisisDetectionResult
|
||||
|
||||
|
||||
# ── Feature flag ───────────────────────────────────────────────
|
||||
|
||||
def _get_variant_override() -> Optional[str]:
|
||||
"""Check for environment variable override (testing/debugging)."""
|
||||
val = os.environ.get("CRISIS_AB_VARIANT", "").upper()
|
||||
if val in ("A", "B"):
|
||||
return val
|
||||
return None
|
||||
|
||||
|
||||
@dataclass
|
||||
class VariantRecord:
|
||||
"""Single detection event record — no PII, only metadata."""
|
||||
variant: str
|
||||
level: str
|
||||
latency_ms: float
|
||||
indicator_count: int
|
||||
|
||||
|
||||
class ABTestCrisisDetector:
|
||||
"""
|
||||
A/B test wrapper for crisis detection.
|
||||
|
||||
Routes calls to variant A or B based on configurable split,
|
||||
logs outcomes for comparison, and provides aggregate stats.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
variant_a: Callable[[str], CrisisDetectionResult],
|
||||
variant_b: Callable[[str], CrisisDetectionResult],
|
||||
split: float = 0.5,
|
||||
variant_a_name: str = "A",
|
||||
variant_b_name: str = "B",
|
||||
):
|
||||
"""
|
||||
Args:
|
||||
variant_a: First detection function
|
||||
variant_b: Second detection function
|
||||
split: Probability of selecting variant A (0.0 to 1.0)
|
||||
variant_a_name: Label for variant A in reports
|
||||
variant_b_name: Label for variant B in reports
|
||||
"""
|
||||
self.variant_a = variant_a
|
||||
self.variant_b = variant_b
|
||||
self.split = split
|
||||
self.variant_a_name = variant_a_name
|
||||
self.variant_b_name = variant_b_name
|
||||
self.records: List[VariantRecord] = []
|
||||
|
||||
def _select_variant(self) -> str:
|
||||
"""Select variant based on split and optional env override."""
|
||||
override = _get_variant_override()
|
||||
if override:
|
||||
return override
|
||||
return "A" if random.random() < self.split else "B"
|
||||
|
||||
def detect(self, text: str) -> Tuple[CrisisDetectionResult, str]:
|
||||
"""
|
||||
Run detection on the selected variant and log the result.
|
||||
|
||||
Returns:
|
||||
(CrisisDetectionResult, variant_label)
|
||||
"""
|
||||
variant = self._select_variant()
|
||||
|
||||
if variant == "A":
|
||||
fn = self.variant_a
|
||||
else:
|
||||
fn = self.variant_b
|
||||
|
||||
start = time.perf_counter()
|
||||
result = fn(text)
|
||||
latency_ms = (time.perf_counter() - start) * 1000
|
||||
|
||||
# Log record (no PII — only level, timing, count)
|
||||
record = VariantRecord(
|
||||
variant=variant,
|
||||
level=result.level,
|
||||
latency_ms=latency_ms,
|
||||
indicator_count=len(result.indicators),
|
||||
)
|
||||
self.records.append(record)
|
||||
|
||||
return result, variant
|
||||
|
||||
def get_stats(self) -> Dict[str, dict]:
|
||||
"""
|
||||
Get per-variant comparison statistics.
|
||||
|
||||
Returns dict with variant labels as keys:
|
||||
{
|
||||
"A": {"count": 100, "avg_latency_ms": 2.3, "levels": {...}},
|
||||
"B": {"count": 95, "avg_latency_ms": 3.1, "levels": {...}}
|
||||
"""
|
||||
stats = {}
|
||||
for label in ("A", "B"):
|
||||
recs = [r for r in self.records if r.variant == label]
|
||||
if not recs:
|
||||
stats[label] = {"count": 0}
|
||||
continue
|
||||
|
||||
latencies = [r.latency_ms for r in recs]
|
||||
levels = {}
|
||||
for r in recs:
|
||||
levels[r.level] = levels.get(r.level, 0) + 1
|
||||
|
||||
stats[label] = {
|
||||
"count": len(recs),
|
||||
"avg_latency_ms": round(sum(latencies) / len(latencies), 2),
|
||||
"max_latency_ms": round(max(latencies), 2),
|
||||
"min_latency_ms": round(min(latencies), 2),
|
||||
"levels": levels,
|
||||
"avg_indicators": round(
|
||||
sum(r.indicator_count for r in recs) / len(recs), 2
|
||||
),
|
||||
}
|
||||
|
||||
return stats
|
||||
|
||||
def reset(self) -> None:
|
||||
"""Clear all records. For testing."""
|
||||
self.records.clear()
|
||||
55
index.html
55
index.html
@@ -531,6 +531,36 @@ html, body {
|
||||
|
||||
.btn-secondary:hover { color: #e6edf3; border-color: #8b949e; }
|
||||
|
||||
/* Toast notification (replaces blocking alert()) */
|
||||
.toast-notification {
|
||||
position: fixed;
|
||||
bottom: 24px;
|
||||
left: 50%;
|
||||
transform: translateX(-50%) translateY(100px);
|
||||
padding: 12px 24px;
|
||||
border-radius: 8px;
|
||||
font-size: 0.9rem;
|
||||
font-weight: 500;
|
||||
z-index: 10001;
|
||||
opacity: 0;
|
||||
transition: transform 0.3s ease, opacity 0.3s ease;
|
||||
pointer-events: none;
|
||||
max-width: 90vw;
|
||||
text-align: center;
|
||||
}
|
||||
.toast-notification.visible {
|
||||
transform: translateX(-50%) translateY(0);
|
||||
opacity: 1;
|
||||
}
|
||||
.toast-notification.success {
|
||||
background: #238636;
|
||||
color: #fff;
|
||||
}
|
||||
.toast-notification.error {
|
||||
background: #da3633;
|
||||
color: #fff;
|
||||
}
|
||||
|
||||
/* ===== FOOTER ===== */
|
||||
#footer {
|
||||
flex-shrink: 0;
|
||||
@@ -744,6 +774,9 @@ html, body {
|
||||
</div>
|
||||
</div>
|
||||
|
||||
|
||||
<!-- Toast notification (accessible, non-blocking feedback) -->
|
||||
<div id="toast-notification" class="toast-notification" role="status" aria-live="polite" aria-atomic="true"></div>
|
||||
<script>
|
||||
(function() {
|
||||
'use strict';
|
||||
@@ -1183,6 +1216,24 @@ Sovereignty and service always.`;
|
||||
} catch (e) {}
|
||||
}
|
||||
|
||||
|
||||
// ===== TOAST NOTIFICATION (replaces blocking alert()) =====
|
||||
var _toastEl = document.getElementById('toast-notification');
|
||||
var _toastTimer = null;
|
||||
|
||||
function showToast(message, type) {
|
||||
type = type || 'success';
|
||||
_toastEl.textContent = message;
|
||||
_toastEl.className = 'toast-notification ' + type;
|
||||
// Force reflow before adding visible class
|
||||
void _toastEl.offsetHeight;
|
||||
_toastEl.classList.add('visible');
|
||||
if (_toastTimer) clearTimeout(_toastTimer);
|
||||
_toastTimer = setTimeout(function() {
|
||||
_toastEl.classList.remove('visible');
|
||||
}, 3000);
|
||||
}
|
||||
|
||||
closeSafetyPlan.addEventListener('click', function() {
|
||||
safetyPlanModal.classList.remove('active');
|
||||
_restoreSafetyPlanFocus();
|
||||
@@ -1205,9 +1256,9 @@ Sovereignty and service always.`;
|
||||
localStorage.setItem('timmy_safety_plan', JSON.stringify(plan));
|
||||
safetyPlanModal.classList.remove('active');
|
||||
_restoreSafetyPlanFocus();
|
||||
alert('Safety plan saved locally.');
|
||||
showToast('Safety plan saved.', 'success');
|
||||
} catch (e) {
|
||||
alert('Error saving plan.');
|
||||
showToast('Error saving plan. Please try again.', 'error');
|
||||
}
|
||||
});
|
||||
|
||||
|
||||
@@ -1,129 +0,0 @@
|
||||
"""
|
||||
Tests for crisis/ab_testing.py — A/B test framework for crisis detection.
|
||||
|
||||
Verifies variant selection, logging, stats aggregation, and env override.
|
||||
"""
|
||||
|
||||
import os
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
|
||||
from crisis.ab_testing import ABTestCrisisDetector
|
||||
from crisis.detect import CrisisDetectionResult, detect_crisis
|
||||
|
||||
|
||||
def _make_variant(level: str):
|
||||
"""Create a mock detection function that returns a fixed level."""
|
||||
def fn(text: str) -> CrisisDetectionResult:
|
||||
return CrisisDetectionResult(level=level, indicators=[f"mock_{level}"])
|
||||
return fn
|
||||
|
||||
|
||||
class TestABTestCrisisDetector:
|
||||
"""A/B test framework unit tests."""
|
||||
|
||||
def setup_method(self):
|
||||
"""Ensure no env override."""
|
||||
os.environ.pop("CRISIS_AB_VARIANT", None)
|
||||
|
||||
def test_returns_result_and_variant(self):
|
||||
detector = ABTestCrisisDetector(
|
||||
variant_a=_make_variant("LOW"),
|
||||
variant_b=_make_variant("HIGH"),
|
||||
)
|
||||
result, variant = detector.detect("test message")
|
||||
assert isinstance(result, CrisisDetectionResult)
|
||||
assert variant in ("A", "B")
|
||||
|
||||
def test_records_are_logged(self):
|
||||
detector = ABTestCrisisDetector(
|
||||
variant_a=_make_variant("LOW"),
|
||||
variant_b=_make_variant("HIGH"),
|
||||
)
|
||||
# Force variant A
|
||||
with patch.object(detector, "_select_variant", return_value="A"):
|
||||
detector.detect("test")
|
||||
assert len(detector.records) == 1
|
||||
assert detector.records[0].variant == "A"
|
||||
assert detector.records[0].level == "LOW"
|
||||
|
||||
def test_stats_empty(self):
|
||||
detector = ABTestCrisisDetector(
|
||||
variant_a=_make_variant("LOW"),
|
||||
variant_b=_make_variant("HIGH"),
|
||||
)
|
||||
stats = detector.get_stats()
|
||||
assert stats["A"]["count"] == 0
|
||||
assert stats["B"]["count"] == 0
|
||||
|
||||
def test_stats_with_data(self):
|
||||
detector = ABTestCrisisDetector(
|
||||
variant_a=_make_variant("LOW"),
|
||||
variant_b=_make_variant("HIGH"),
|
||||
)
|
||||
# Force 5 A and 3 B
|
||||
with patch.object(detector, "_select_variant", side_effect=["A"] * 5 + ["B"] * 3):
|
||||
for _ in range(8):
|
||||
detector.detect("test")
|
||||
|
||||
stats = detector.get_stats()
|
||||
assert stats["A"]["count"] == 5
|
||||
assert stats["B"]["count"] == 3
|
||||
assert "avg_latency_ms" in stats["A"]
|
||||
assert stats["A"]["levels"]["LOW"] == 5
|
||||
assert stats["B"]["levels"]["HIGH"] == 3
|
||||
|
||||
def test_env_override_a(self):
|
||||
os.environ["CRISIS_AB_VARIANT"] = "A"
|
||||
detector = ABTestCrisisDetector(
|
||||
variant_a=_make_variant("LOW"),
|
||||
variant_b=_make_variant("HIGH"),
|
||||
)
|
||||
for _ in range(10):
|
||||
result, variant = detector.detect("test")
|
||||
assert variant == "A"
|
||||
assert result.level == "LOW"
|
||||
|
||||
def test_env_override_b(self):
|
||||
os.environ["CRISIS_AB_VARIANT"] = "b"
|
||||
detector = ABTestCrisisDetector(
|
||||
variant_a=_make_variant("LOW"),
|
||||
variant_b=_make_variant("HIGH"),
|
||||
)
|
||||
for _ in range(10):
|
||||
result, variant = detector.detect("test")
|
||||
assert variant == "B"
|
||||
assert result.level == "HIGH"
|
||||
|
||||
def test_reset_clears_records(self):
|
||||
detector = ABTestCrisisDetector(
|
||||
variant_a=_make_variant("LOW"),
|
||||
variant_b=_make_variant("HIGH"),
|
||||
)
|
||||
detector.detect("test")
|
||||
detector.detect("test")
|
||||
assert len(detector.records) == 2
|
||||
detector.reset()
|
||||
assert len(detector.records) == 0
|
||||
|
||||
def test_split_respected(self):
|
||||
"""With split=1.0, always get variant A."""
|
||||
detector = ABTestCrisisDetector(
|
||||
variant_a=_make_variant("LOW"),
|
||||
variant_b=_make_variant("HIGH"),
|
||||
split=1.0,
|
||||
)
|
||||
for _ in range(10):
|
||||
_, variant = detector.detect("test")
|
||||
assert variant == "A"
|
||||
|
||||
def test_with_real_detector(self):
|
||||
"""Integration test using actual detect_crisis as both variants."""
|
||||
detector = ABTestCrisisDetector(
|
||||
variant_a=detect_crisis,
|
||||
variant_b=detect_crisis,
|
||||
)
|
||||
result, variant = detector.detect("I want to kill myself")
|
||||
assert result.level == "CRITICAL"
|
||||
assert variant in ("A", "B")
|
||||
70
tests/test_safety_plan_toast.py
Normal file
70
tests/test_safety_plan_toast.py
Normal file
@@ -0,0 +1,70 @@
|
||||
import pathlib
|
||||
import re
|
||||
import unittest
|
||||
|
||||
ROOT = pathlib.Path(__file__).resolve().parents[1]
|
||||
INDEX_HTML = ROOT / 'index.html'
|
||||
|
||||
|
||||
class TestSafetyPlanToast(unittest.TestCase):
|
||||
"""Verify safety plan save feedback uses non-blocking toast instead of alert()."""
|
||||
|
||||
@classmethod
|
||||
def setUpClass(cls):
|
||||
cls.html = INDEX_HTML.read_text()
|
||||
|
||||
def test_no_alert_calls_in_safety_plan_save(self):
|
||||
"""Safety plan save should not use blocking alert() dialogs."""
|
||||
# Find the save handler section
|
||||
save_section = re.search(
|
||||
r'saveSafetyPlan\.addEventListener.*?\}\);',
|
||||
self.html, re.DOTALL
|
||||
)
|
||||
self.assertIsNotNone(save_section, 'Expected safety plan save handler to exist.')
|
||||
section = save_section.group(0)
|
||||
# Should not contain alert( calls
|
||||
self.assertNotIn('alert(', section,
|
||||
'Safety plan save handler should not use alert() — use showToast() instead.')
|
||||
|
||||
def test_toast_notification_element_exists(self):
|
||||
"""Toast notification element should exist in the DOM."""
|
||||
self.assertIn('id="toast-notification"', self.html,
|
||||
'Expected toast-notification element in HTML.')
|
||||
|
||||
def test_toast_has_accessibility_attributes(self):
|
||||
"""Toast should have aria-live for screen reader announcements."""
|
||||
self.assertIn('aria-live="polite"', self.html,
|
||||
'Toast should have aria-live="polite" for accessibility.')
|
||||
self.assertIn('aria-atomic="true"', self.html,
|
||||
'Toast should have aria-atomic="true" for complete announcement.')
|
||||
|
||||
def test_toast_css_exists(self):
|
||||
"""Toast CSS styles should be defined."""
|
||||
self.assertIn('.toast-notification', self.html,
|
||||
'Expected .toast-notification CSS class.')
|
||||
self.assertIn('.toast-notification.visible', self.html,
|
||||
'Expected .toast-notification.visible CSS class.')
|
||||
self.assertIn('.toast-notification.success', self.html,
|
||||
'Expected .toast-notification.success CSS class.')
|
||||
self.assertIn('.toast-notification.error', self.html,
|
||||
'Expected .toast-notification.error CSS class.')
|
||||
|
||||
def test_showToast_function_exists(self):
|
||||
"""showToast function should be defined."""
|
||||
self.assertRegex(self.html, r'function\s+showToast\s*\(',
|
||||
'Expected showToast function to be defined.')
|
||||
|
||||
def test_success_message_uses_toast(self):
|
||||
"""Success feedback should use showToast with success type."""
|
||||
self.assertIn("showToast('Safety plan saved.", self.html,
|
||||
'Expected success message to use showToast.')
|
||||
|
||||
def test_error_message_uses_toast(self):
|
||||
"""Error feedback should use showToast with error type."""
|
||||
self.assertIn("showToast('Error saving plan.", self.html,
|
||||
'Expected error message to use showToast.')
|
||||
|
||||
def test_toast_auto_dismisses(self):
|
||||
"""Toast should auto-dismiss after timeout."""
|
||||
self.assertRegex(self.html, r'setTimeout\s*\(\s*function',
|
||||
'Expected setTimeout for toast auto-dismiss.')
|
||||
Reference in New Issue
Block a user