test: add A/B testing framework tests (#101)

GENOME.md

@@ -1,75 +0,0 @@
-# GENOME.md — the-door
-
-**Generated:** 2026-04-14
-**Repo:** Timmy_Foundation/the-door
-**Description:** Crisis Front Door — a single URL where a man at 3am can talk to Timmy. No login, no signup. 988 always visible.
-
----
-
-## Project Overview
-
-The-door is a crisis intervention web application — the most sacred surface in the Timmy Foundation. When a man at 3am reaches the end of his road, this is where he lands. No login, no signup, no barriers. 988 Suicide and Crisis Lifeline always visible. The "When a Man Is Dying" protocol active on every page.
-
-## Architecture
-
-```
-the-door/
-├── index.html              # Main crisis page (PWA-capable)
-├── crisis-offline.html     # Offline fallback (service worker cached)
-├── about.html              # About page
-├── testimony.html          # Testimony/stories page
-├── sw.js                   # Service worker (offline-first)
-├── manifest.json           # PWA manifest
-├── crisis/                 # Core crisis detection + response
-│   ├── detect.py           # Keyword/pattern detection (4 tiers)
-│   ├── gateway.py          # API endpoints, prompt injection
-│   ├── response.py         # Response generation, 988 routing
-│   ├── compassion_router.py # Profile-based response routing
-│   ├── profiles.py         # Compassion profiles
-│   └── PROTOCOL.md         # The protocol (SOUL.md reference)
-├── crisis_detector.py      # Legacy shim → crisis/detect.py
-├── crisis_responder.py     # Legacy responder
-├── dying_detection/        # Deprecated module
-├── evolution/              # Crisis synthesizer (creative)
-├── tests/                  # Safety-critical tests
-│   ├── test_crisis_overlay_focus_trap.py
-│   ├── test_dying_detection_deprecation.py
-│   └── test_false_positive_fixes.py
-└── deploy/                 # Deployment docs
-```
-
-## Key Abstractions
-
-| Module | Purpose |
-|---|---|
-| `crisis/detect.py` | 4-tier detection: LOW/MEDIUM/HIGH/CRITICAL via regex patterns |
-| `crisis/gateway.py` | HTTP API, Sovereign Heart prompt injection |
-| `crisis/response.py` | Response generation, 988 integration, escalation |
-| `crisis/compassion_router.py` | Profile-based routing (different crisis types) |
-| `sw.js` | Service worker for offline-first PWA |
-
-## Safety Constraints
-
-- **The-door never auto-closes PRs** (in fleet-ops exempt list)
-- **988 always visible** on every page, even offline
-- **When a Man Is Dying protocol** active on every interaction
-- **No login/signup** — zero barriers to crisis support
-- **Offline-first** — service worker caches critical pages
-
-## Test Coverage
-
-| Test | Coverage |
-|---|---|
-| Crisis overlay focus trap | ✅ |
-| Dying detection deprecation | ✅ |
-| False positive fixes | ✅ |
-| Crisis detection tiers | ❌ (in crisis/tests.py) |
-| Response generation | ❌ |
-| Offline service worker | ❌ |
-
-## Security
-
-- No user data stored (crisis intervention is stateless by design)
-- No cookies, no tracking, no analytics
-- Service worker only caches static assets
-- Crisis detection runs client-side where possible

crisis/ab_testing.py

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

tests/test_ab_testing.py

@@ -0,0 +1,129 @@
+"""
+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")