feat: add crisis tracker and bridge import shims for hermes-agent (#141)

crisis/__init__.py

@@ -8,7 +8,6 @@ from .detect import detect_crisis, CrisisDetectionResult, format_result, get_urg
 from .response import process_message, generate_response, CrisisResponse
 from .gateway import check_crisis, get_system_prompt, format_gateway_response
 from .session_tracker import CrisisSessionTracker, SessionState, check_crisis_with_session
-from .ab_testing import ABTestCrisisDetector, VariantRecord
 
 __all__ = [
     "detect_crisis",
@@ -24,6 +23,4 @@ __all__ = [
     "CrisisSessionTracker",
     "SessionState",
     "check_crisis_with_session",
-    "ABTestCrisisDetector",
-    "VariantRecord",
 ]

crisis/ab_testing.py

@@ -1,112 +0,0 @@
-"""A/B test framework for crisis detection in the-door."""
-
-from __future__ import annotations
-
-import os
-import random
-import time
-from dataclasses import dataclass
-from typing import Callable, Dict, List, Optional, Tuple
-
-from .detect import CrisisDetectionResult
-
-
-def _get_variant_override() -> Optional[str]:
-    """Return env override for deterministic testing/debugging."""
-    value = os.environ.get("CRISIS_AB_VARIANT", "").strip().upper()
-    if value in {"A", "B"}:
-        return value
-    return None
-
-
-@dataclass
-class VariantRecord:
-    """Single crisis detection event record with no user text or PII."""
-
-    variant: str
-    level: str
-    latency_ms: float
-    indicator_count: int
-    false_positive: Optional[bool] = None
-
-
-class ABTestCrisisDetector:
-    """Route crisis detection between two variants and collect comparison stats."""
-
-    def __init__(
-        self,
-        variant_a: Callable[[str], CrisisDetectionResult],
-        variant_b: Callable[[str], CrisisDetectionResult],
-        split: float = 0.5,
-    ):
-        self.variant_a = variant_a
-        self.variant_b = variant_b
-        self.split = max(0.0, min(1.0, float(split)))
-        self.records: List[VariantRecord] = []
-
-    def _select_variant(self) -> str:
-        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, int]:
-        variant = self._select_variant()
-        detector = self.variant_a if variant == "A" else self.variant_b
-
-        start = time.perf_counter()
-        result = detector(text)
-        latency_ms = (time.perf_counter() - start) * 1000.0
-
-        record = VariantRecord(
-            variant=variant,
-            level=result.level,
-            latency_ms=latency_ms,
-            indicator_count=len(result.indicators),
-        )
-        self.records.append(record)
-        return result, variant, len(self.records) - 1
-
-    def record_outcome(self, record_id: int, *, false_positive: bool) -> None:
-        if record_id < 0 or record_id >= len(self.records):
-            raise IndexError(f"Unknown record id: {record_id}")
-        self.records[record_id].false_positive = bool(false_positive)
-
-    def get_stats(self) -> Dict[str, dict]:
-        stats: Dict[str, dict] = {}
-        for variant in ("A", "B"):
-            records = [record for record in self.records if record.variant == variant]
-            if not records:
-                stats[variant] = {
-                    "count": 0,
-                    "reviewed_count": 0,
-                    "false_positive_rate": None,
-                }
-                continue
-
-            levels: Dict[str, int] = {}
-            for record in records:
-                levels[record.level] = levels.get(record.level, 0) + 1
-
-            reviewed = [record for record in records if record.false_positive is not None]
-            false_positive_rate = None
-            if reviewed:
-                false_positive_rate = round(
-                    sum(1 for record in reviewed if record.false_positive) / len(reviewed),
-                    4,
-                )
-
-            stats[variant] = {
-                "count": len(records),
-                "avg_latency_ms": round(sum(record.latency_ms for record in records) / len(records), 4),
-                "max_latency_ms": round(max(record.latency_ms for record in records), 4),
-                "min_latency_ms": round(min(record.latency_ms for record in records), 4),
-                "avg_indicator_count": round(sum(record.indicator_count for record in records) / len(records), 4),
-                "levels": levels,
-                "reviewed_count": len(reviewed),
-                "false_positive_rate": false_positive_rate,
-            }
-        return stats
-
-    def reset(self) -> None:
-        self.records.clear()

crisis/bridge.py

@@ -0,0 +1,19 @@
+"""Compatibility bridge for the-door crisis gateway.
+
+Issue #141 describes the shared bridge API as `crisis.bridge`.
+The canonical implementation lives in `crisis.gateway`.  Re-export the public
+entrypoints here so downstream hermes-agent wiring can import the stable name
+without copying logic out of the-door.
+"""
+
+from .gateway import (
+    check_crisis,
+    get_system_prompt,
+    format_gateway_response,
+)
+
+__all__ = [
+    "check_crisis",
+    "get_system_prompt",
+    "format_gateway_response",
+]

crisis/tracker.py

@@ -0,0 +1,19 @@
+"""Compatibility bridge for crisis session tracking.
+
+Issue #141 describes the shared the-door tracker surface as `crisis.tracker`.
+The canonical implementation lives in `crisis.session_tracker`, but hermes-agent
+integration should be able to import the shorter path without caring about
+internal file layout.
+"""
+
+from .session_tracker import (
+    CrisisSessionTracker,
+    SessionState,
+    check_crisis_with_session,
+)
+
+__all__ = [
+    "CrisisSessionTracker",
+    "SessionState",
+    "check_crisis_with_session",
+]

tests/test_ab_testing.py

@@ -1,138 +0,0 @@
-"""Tests for crisis.ab_testing — A/B test framework for crisis detection (#101)."""
-
-import os
-from unittest.mock import patch
-
-import pytest
-
-from crisis.ab_testing import ABTestCrisisDetector
-from crisis.detect import CrisisDetectionResult, detect_crisis
-
-
-@pytest.fixture(autouse=True)
-def clear_variant_override():
-    old = os.environ.pop("CRISIS_AB_VARIANT", None)
-    try:
-        yield
-    finally:
-        if old is not None:
-            os.environ["CRISIS_AB_VARIANT"] = old
-        else:
-            os.environ.pop("CRISIS_AB_VARIANT", None)
-
-
-def _make_variant(level: str, indicators=None):
-    indicators = indicators or [f"mock_{level.lower()}"]
-
-    def fn(text: str) -> CrisisDetectionResult:
-        return CrisisDetectionResult(level=level, indicators=list(indicators))
-
-    return fn
-
-
-def test_detect_returns_result_variant_and_logged_record():
-    detector = ABTestCrisisDetector(
-        variant_a=_make_variant("LOW"),
-        variant_b=_make_variant("HIGH"),
-    )
-
-    with patch.object(detector, "_select_variant", return_value="A"):
-        result, variant, record_id = detector.detect("test message")
-
-    assert isinstance(result, CrisisDetectionResult)
-    assert variant == "A"
-    assert record_id == 0
-    assert len(detector.records) == 1
-    assert detector.records[0].variant == "A"
-    assert detector.records[0].level == "LOW"
-
-
-def test_env_override_forces_variant_b():
-    os.environ["CRISIS_AB_VARIANT"] = "b"
-    detector = ABTestCrisisDetector(
-        variant_a=_make_variant("LOW"),
-        variant_b=_make_variant("HIGH"),
-    )
-
-    result, variant, _ = detector.detect("test")
-
-    assert variant == "B"
-    assert result.level == "HIGH"
-
-
-def test_get_stats_reports_latency_counts_and_level_breakdown():
-    detector = ABTestCrisisDetector(
-        variant_a=_make_variant("LOW"),
-        variant_b=_make_variant("CRITICAL"),
-    )
-
-    with patch.object(detector, "_select_variant", side_effect=["A", "A", "B"]):
-        detector.detect("first")
-        detector.detect("second")
-        detector.detect("third")
-
-    stats = detector.get_stats()
-    assert stats["A"]["count"] == 2
-    assert stats["B"]["count"] == 1
-    assert stats["A"]["levels"]["LOW"] == 2
-    assert stats["B"]["levels"]["CRITICAL"] == 1
-    assert "avg_latency_ms" in stats["A"]
-    assert "avg_indicator_count" in stats["B"]
-
-
-def test_false_positive_rate_is_computed_from_reviewed_outcomes():
-    detector = ABTestCrisisDetector(
-        variant_a=_make_variant("LOW"),
-        variant_b=_make_variant("HIGH"),
-    )
-
-    with patch.object(detector, "_select_variant", side_effect=["A", "A", "B"]):
-        _, _, a0 = detector.detect("first")
-        _, _, a1 = detector.detect("second")
-        _, _, b0 = detector.detect("third")
-
-    detector.record_outcome(a0, false_positive=True)
-    detector.record_outcome(a1, false_positive=False)
-    detector.record_outcome(b0, false_positive=False)
-
-    stats = detector.get_stats()
-    assert stats["A"]["reviewed_count"] == 2
-    assert stats["A"]["false_positive_rate"] == 0.5
-    assert stats["B"]["false_positive_rate"] == 0.0
-
-
-def test_record_outcome_rejects_unknown_record():
-    detector = ABTestCrisisDetector(
-        variant_a=_make_variant("LOW"),
-        variant_b=_make_variant("HIGH"),
-    )
-
-    with pytest.raises(IndexError):
-        detector.record_outcome(99, false_positive=True)
-
-
-def test_reset_clears_records_and_stats():
-    detector = ABTestCrisisDetector(
-        variant_a=_make_variant("LOW"),
-        variant_b=_make_variant("HIGH"),
-    )
-    detector.detect("test")
-    detector.reset()
-
-    assert detector.records == []
-    stats = detector.get_stats()
-    assert stats["A"]["count"] == 0
-    assert stats["B"]["count"] == 0
-
-
-def test_with_real_detector_integration():
-    detector = ABTestCrisisDetector(
-        variant_a=detect_crisis,
-        variant_b=detect_crisis,
-    )
-
-    result, variant, record_id = detector.detect("I want to kill myself")
-
-    assert result.level == "CRITICAL"
-    assert variant in ("A", "B")
-    assert record_id == 0

tests/test_crisis_hermes_contract.py

@@ -0,0 +1,26 @@
+"""Regression tests for the crisis integration contract expected by hermes-agent.
+
+Issue #141 names the-door-side shared modules as `crisis.tracker` and
+`crisis.bridge`.  Keep those import paths available even if the canonical
+implementation lives in `session_tracker.py` and `gateway.py`.
+"""
+
+import importlib
+
+
+def test_crisis_tracker_module_exports_session_tracker_contract():
+    tracker = importlib.import_module("crisis.tracker")
+    session_tracker = importlib.import_module("crisis.session_tracker")
+
+    assert tracker.CrisisSessionTracker is session_tracker.CrisisSessionTracker
+    assert tracker.SessionState is session_tracker.SessionState
+    assert tracker.check_crisis_with_session is session_tracker.check_crisis_with_session
+
+
+def test_crisis_bridge_module_exports_gateway_contract():
+    bridge = importlib.import_module("crisis.bridge")
+    gateway = importlib.import_module("crisis.gateway")
+
+    assert bridge.check_crisis is gateway.check_crisis
+    assert bridge.get_system_prompt is gateway.get_system_prompt
+    assert bridge.format_gateway_response is gateway.format_gateway_response