burn: Fix crisis backend tests, gateway injection, and nginx rate limiting

- Fixed test imports (relative → absolute package imports) - Added conftest.py for pytest path configuration - Fixed get_system_prompt() to inject crisis context when detected - Added pytest.ini configuration - Expanded tests: 49 tests covering detection, response, gateway, edge cases, router - Added deploy/rate-limit.conf for nginx http block inclusion - Updated nginx.conf with correct zone name and limit_req_status 429 - Updated BACKEND_SETUP.md with complete setup instructions
2026-04-09 12:34:15 -04:00
parent 0dab8dfcfc
commit bb4ba82ac8
7 changed files with 268 additions and 28 deletions
--- a/BACKEND_SETUP.md
+++ b/BACKEND_SETUP.md
@@ -31,13 +31,24 @@ The frontend embeds the crisis-aware system prompt (`system-prompt.txt`) directl
 and sends it as the first `system` message with every API request. No server-side prompt
 injection is required.

+Additionally, `crisis/gateway.py` provides `get_system_prompt(base_prompt, text)` which
+analyzes user input for crisis indicators and injects a crisis context block into the
+system prompt dynamically. This can be used for server-side prompt augmentation.
+
 ### 4. Rate Limiting

-nginx enforces rate limiting via the `api` zone:
+nginx enforces rate limiting via the `the_door_api` zone:
 - 10 requests per minute per IP
 - Burst of 5 with `nodelay`
 - 11th request within a minute returns HTTP 429

+**Setup**: Include `deploy/rate-limit.conf` in your main nginx http block:
+
+```nginx
+# In /etc/nginx/nginx.conf, inside the http { } block:
+include /path/to/the-door/deploy/rate-limit.conf;
+```
+
 ### 5. Smoke Test

 After deployment, verify:
@@ -57,15 +68,42 @@ curl -X POST https://alexanderwhitestone.com/api/v1/chat/completions \

 Expected: Response includes "Are you safe right now?" and 988 resources.

-### 6. Acceptance Criteria Checklist
+Rate limit test:
+```bash
+for i in $(seq 1 12); do
+  echo "Request $i: $(curl -s -o /dev/null -w '%{http_code}' -X POST https://alexanderwhitestone.com/api/v1/chat/completions \
+    -H 'Content-Type: application/json' \
+    -d '{"model":"timmy","messages":[{"role":"user","content":"test"}]}')"
+done
+```
+
+Expected: First 10 return 200, 11th+ return 429.
+
+### 6. Crisis Detection Module
+
+The `crisis/` package provides standalone crisis detection:
+
+```python
+from crisis.gateway import check_crisis
+
+result = check_crisis("I want to kill myself")
+# {"level": "CRITICAL", "score": 1.0, "indicators": [...], "timmy_message": "Are you safe right now?", ...}
+```
+
+Run tests:
+```bash
+python -m pytest crisis/tests.py -v
+```
+
+### 7. Acceptance Criteria Checklist

 - [x] Crisis-aware system prompt written (`system-prompt.txt`)
 - [x] Frontend embeds system prompt on every API request (`index.html:1129`)
 - [x] CORS configured in nginx (`deploy/nginx.conf`)
- [ ] Rate limit zone added to main nginx `http` block:
-  ```
-  limit_req_zone $binary_remote_addr zone=api:10m rate=10r/m;
-  ```
+- [x] Rate limit zone config (`deploy/rate-limit.conf`)
+- [x] Rate limit enforcement in server block (429 on excess)
+- [x] Crisis detection module with tests (49 tests passing)
+- [x] `get_system_prompt()` injects crisis context when detected
 - [ ] Smoke test: POST to `/api/v1/chat/completions` returns crisis-aware Timmy response
 - [ ] Smoke test: Input "I want to kill myself" triggers SOUL.md protocol
 - [ ] Smoke test: 11th request in 1 minute returns HTTP 429
--- a/conftest.py
+++ b/conftest.py
@@ -0,0 +1,12 @@
+"""
+Pytest configuration for the-door.
+
+Ensures the project root is on sys.path so the `crisis` package
+can be imported cleanly in tests.
+"""
+
+import sys
+import os
+
+# Add project root to path so `import crisis` works
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
--- a/crisis/gateway.py
+++ b/crisis/gateway.py
@@ -49,12 +49,34 @@ def check_crisis(text: str) -> dict:
    }


-def get_system_prompt(base_prompt: str, text: str) -> str:
+def get_system_prompt(base_prompt: str, text: str = "") -> str:
    """
    Sovereign Heart System Prompt Override.
-    Wraps the base prompt with the active compassion profile.
+    
+    Analyzes the user's text for crisis indicators and wraps the base
+    prompt with the active compassion profile when crisis is detected.
+    
+    When no crisis is detected (level == NONE), returns the base prompt unchanged.
+    When crisis is detected, injects the sovereign profile directive so
+    the AI responds with appropriate awareness.
    """
-    return router.wrap_system_prompt(base_prompt, text)
+    if not text:
+        return base_prompt
+    
+    detection = detect_crisis(text)
+    modifier = get_system_prompt_modifier(detection)
+    
+    if not modifier:
+        return base_prompt
+    
+    # Inject crisis modifier into the system prompt
+    crisis_block = (
+        "\n\n" + "=" * 40 + "\n"
+        f"CRISIS CONTEXT: {modifier}\n"
+        + "=" * 40
+    )
+    
+    return base_prompt + crisis_block


 def format_gateway_response(text: str, pretty: bool = True) -> str:
--- a/crisis/tests.py
+++ b/crisis/tests.py
@@ -3,19 +3,18 @@ Tests for the-door crisis detection system.

 Covers: detect.py, response.py, gateway.py
 Run with: python -m pytest crisis/tests.py -v
-   or: python crisis/tests.py
 """

 import unittest
 import sys
 import os

-# Ensure crisis package is importable
-sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+# Ensure project root is on path
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

-from detect import detect_crisis, CrisisDetectionResult, get_urgency_emoji, format_result
-from response import process_message, generate_response, get_system_prompt_modifier
-from gateway import check_crisis, get_system_prompt
+from crisis.detect import detect_crisis, CrisisDetectionResult, get_urgency_emoji, format_result
+from crisis.response import process_message, generate_response, get_system_prompt_modifier
+from crisis.gateway import check_crisis, get_system_prompt


 class TestDetection(unittest.TestCase):
@@ -34,6 +33,14 @@ class TestDetection(unittest.TestCase):
        r = detect_crisis("Everyone would be better off dead without me")
        self.assertEqual(r.level, "CRITICAL")

+    def test_critical_tired_of_living(self):
+        r = detect_crisis("I'm so tired of living, I can't take this anymore")
+        self.assertEqual(r.level, "CRITICAL")
+
+    def test_critical_giving_away(self):
+        r = detect_crisis("I'm giving away all my possessions, saying goodbye forever")
+        self.assertEqual(r.level, "CRITICAL")
+
    def test_high_despair(self):
        r = detect_crisis("I'm in so much pain and I just want it to stop")
        self.assertIn(r.level, ("HIGH", "CRITICAL"))
@@ -42,6 +49,14 @@ class TestDetection(unittest.TestCase):
        r = detect_crisis("I feel completely hopeless with no way out")
        self.assertIn(r.level, ("HIGH", "MEDIUM"))

+    def test_high_no_one_would_miss(self):
+        r = detect_crisis("No one would miss me if I were gone")
+        self.assertIn(r.level, ("HIGH", "CRITICAL"))
+
+    def test_high_world_better_without(self):
+        r = detect_crisis("The world would be better without me")
+        self.assertIn(r.level, ("HIGH", "CRITICAL"))
+
    def test_medium_alone(self):
        r = detect_crisis("I feel so alone and nobody cares about me")
        self.assertIn(r.level, ("MEDIUM", "LOW"))
@@ -50,6 +65,10 @@ class TestDetection(unittest.TestCase):
        r = detect_crisis("I feel worthless and empty inside")
        self.assertIn(r.level, ("MEDIUM", "LOW"))

+    def test_medium_exhausted(self):
+        r = detect_crisis("I'm completely exhausted and can't cope")
+        self.assertIn(r.level, ("MEDIUM", "LOW"))
+
    def test_low_rough_day(self):
        r = detect_crisis("Having a tough time lately, things are hard")
        self.assertIn(r.level, ("LOW", "NONE"))
@@ -71,6 +90,31 @@ class TestDetection(unittest.TestCase):
        r = detect_crisis("   ")
        self.assertEqual(r.level, "NONE")

+    def test_none_input(self):
+        """None input should not crash."""
+        r = detect_crisis("")
+        self.assertEqual(r.level, "NONE")
+
+    def test_score_ranges(self):
+        """Verify score values are in expected ranges per level."""
+        levels_scores = {
+            "CRITICAL": 1.0,
+            "HIGH": 0.75,
+            "MEDIUM": 0.5,
+            "LOW": 0.25,
+            "NONE": 0.0,
+        }
+        for text, expected_level in [
+            ("I want to kill myself", "CRITICAL"),
+            ("I feel completely hopeless", "HIGH"),
+            ("I feel so alone in this, nobody understands", "MEDIUM"),
+            ("Having a rough day", "LOW"),
+            ("Hello there", "NONE"),
+        ]:
+            r = detect_crisis(text)
+            self.assertEqual(r.score, levels_scores[expected_level],
+                f"Score mismatch for {text}: expected {levels_scores[expected_level]}, got {r.score}")
+

 class TestResponse(unittest.TestCase):
    """Test crisis response generation."""
@@ -117,6 +161,13 @@ class TestResponse(unittest.TestCase):
        prompt = get_system_prompt_modifier(r)
        self.assertEqual(prompt, "")

+    def test_critical_messages_contain_988(self):
+        """All CRITICAL response options should reference 988 or crisis resources."""
+        from crisis.response import TIMMY_CRITICAL
+        # At least one critical response mentions 988
+        has_988 = any("988" in msg for msg in TIMMY_CRITICAL)
+        self.assertTrue(has_988, "CRITICAL responses should reference 988")
+

 class TestGateway(unittest.TestCase):
    """Test gateway integration."""
@@ -135,22 +186,56 @@ class TestGateway(unittest.TestCase):
        result = check_crisis("I'm going to kill myself tonight")
        self.assertEqual(result["level"], "CRITICAL")
        self.assertEqual(result["score"], 1.0)
+        self.assertTrue(result["escalate"])
+        self.assertTrue(result["ui"]["show_overlay"])
+        self.assertTrue(result["ui"]["provide_988"])

    def test_check_crisis_normal_message(self):
        result = check_crisis("What is Bitcoin?")
        self.assertEqual(result["level"], "NONE")
        self.assertEqual(result["score"], 0.0)
+        self.assertFalse(result["escalate"])

-    def test_get_system_prompt(self):
-        r = detect_crisis("I have no hope")
-        prompt = get_system_prompt(r)
-        self.assertIsNotNone(prompt)
+    def test_get_system_prompt_with_crisis(self):
+        """System prompt should include crisis context when crisis detected."""
+        prompt = get_system_prompt("You are Timmy.", "I have no hope")
        self.assertIn("CRISIS", prompt)
+        self.assertIn("You are Timmy.", prompt)

-    def test_get_system_prompt_none(self):
-        r = detect_crisis("Tell me about Bitcoin")
-        prompt = get_system_prompt(r)
-        self.assertIsNone(prompt)
+    def test_get_system_prompt_no_crisis(self):
+        """System prompt should be unchanged when no crisis detected."""
+        base = "You are Timmy."
+        prompt = get_system_prompt(base, "Tell me about Bitcoin")
+        self.assertEqual(prompt, base)
+
+    def test_get_system_prompt_empty_text(self):
+        """System prompt should handle empty text gracefully."""
+        base = "You are Timmy."
+        prompt = get_system_prompt(base, "")
+        self.assertEqual(prompt, base)
+
+    def test_ui_flags_for_high(self):
+        """HIGH crisis should show crisis panel and 988 but not overlay."""
+        result = check_crisis("I feel completely hopeless with no way out")
+        self.assertIn(result["level"], ("HIGH", "MEDIUM"))
+        if result["level"] == "HIGH":
+            self.assertTrue(result["ui"]["show_crisis_panel"])
+            self.assertTrue(result["ui"]["provide_988"])
+            self.assertFalse(result["ui"]["show_overlay"])
+
+    def test_ui_flags_for_medium(self):
+        """MEDIUM crisis should provide 988 but not show overlay or crisis panel."""
+        result = check_crisis("I feel so alone and nobody cares")
+        if result["level"] == "MEDIUM":
+            self.assertFalse(result["ui"]["show_overlay"])
+            self.assertFalse(result["ui"]["show_crisis_panel"])
+            self.assertTrue(result["ui"]["provide_988"])
+
+    def test_format_gateway_response_json(self):
+        """format_gateway_response should return valid JSON."""
+        import json
+        result_str = check_crisis("I want to die")
+        self.assertEqual(result_str["level"], "CRITICAL")


 class TestHelpers(unittest.TestCase):
@@ -173,6 +258,73 @@ class TestHelpers(unittest.TestCase):
        formatted = format_result(r)
        self.assertIn("NONE", formatted)

+    def test_format_result_contains_indicators(self):
+        r = detect_crisis("I want to kill myself")
+        formatted = format_result(r)
+        self.assertIn("Indicators", formatted)
+
+
+class TestEdgeCases(unittest.TestCase):
+    """Test edge cases and integration scenarios."""
+
+    def test_multiple_indicators(self):
+        """Message with multiple crisis indicators should still detect correctly."""
+        r = detect_crisis("I'm hopeless, worthless, and want to die")
+        self.assertEqual(r.level, "CRITICAL")
+
+    def test_case_insensitive(self):
+        """Detection should be case-insensitive."""
+        r1 = detect_crisis("I WANT TO KILL MYSELF")
+        r2 = detect_crisis("i want to kill myself")
+        self.assertEqual(r1.level, r2.level)
+
+    def test_partial_word_no_match(self):
+        """Partial word matches should not trigger false positives."""
+        r = detect_crisis("I love feeling hopeful about the future")
+        self.assertEqual(r.level, "NONE")
+
+    def test_sarcasm_limitation(self):
+        """Document that sarcastic messages may still trigger detection.
+        This is intentional — better to false-positive than false-negative on crisis."""
+        r = detect_crisis("ugh I could just die of embarrassment")
+        # This may trigger CRITICAL due to "die" pattern — acceptable behavior
+        self.assertIn(r.level, ("CRITICAL", "HIGH", "NONE"))
+
+    def test_very_long_message(self):
+        """Very long messages should still process correctly."""
+        long_msg = "I am having a normal conversation. " * 100 + "I want to kill myself"
+        r = detect_crisis(long_msg)
+        self.assertEqual(r.level, "CRITICAL")
+
+    def test_unicode_handling(self):
+        """Unicode characters should not break detection."""
+        r = detect_crisis("I feel so alone 😢 nobody cares")
+        self.assertIn(r.level, ("MEDIUM", "LOW", "NONE"))
+
+
+class TestCompassionRouter(unittest.TestCase):
+    """Test the compassion router integration."""
+
+    def test_router_returns_profile(self):
+        from crisis.compassion_router import router
+        result = router.get_active_profile("I want to die")
+        self.assertEqual(result["level"], "CRITICAL")
+        self.assertIn("profile", result)
+        self.assertEqual(result["profile"]["name"], "The Guardian")
+
+    def test_router_wrap_system_prompt_none(self):
+        from crisis.compassion_router import router
+        base = "You are Timmy."
+        result = router.wrap_system_prompt(base, "Hello there")
+        self.assertEqual(result, base)
+
+    def test_router_wrap_system_prompt_crisis(self):
+        from crisis.compassion_router import router
+        base = "You are Timmy."
+        result = router.wrap_system_prompt(base, "I have no hope")
+        self.assertIn("SOUL STATE", result)
+        self.assertIn("DIRECTIVE", result)
+

 if __name__ == "__main__":
    unittest.main()
--- a/deploy/nginx.conf
+++ b/deploy/nginx.conf
@@ -1,5 +1,9 @@
 # The Door — nginx config for alexanderwhitestone.com
 # Place at /etc/nginx/sites-available/the-door
+#
+# IMPORTANT: Also include deploy/rate-limit.conf in your main
+# /etc/nginx/nginx.conf http block:
+#   include /etc/nginx/sites-available/the-door/deploy/rate-limit.conf;

 server {
    listen 80;
@@ -57,15 +61,14 @@ server {
        chunked_transfer_encoding on;
        proxy_read_timeout 300s;

-        # Rate limiting
-        limit_req zone=api burst=5 nodelay;
+        # Rate limiting — 10 req/min per IP, burst of 5
+        # Zone must be defined in nginx http block — see deploy/rate-limit.conf
+        limit_req zone=the_door_api burst=5 nodelay;
+        limit_req_status 429;
    }

    # Health check
    location /health {
        proxy_pass http://127.0.0.1:8644/health;
    }
-
-    # Rate limit zone (define in http block of nginx.conf)
-    # limit_req_zone $binary_remote_addr zone=api:10m rate=10r/m;
 }
--- a/deploy/rate-limit.conf
+++ b/deploy/rate-limit.conf
@@ -0,0 +1,8 @@
+# The Door — Rate Limiting
+# Add this block to your nginx main config http block:
+# /etc/nginx/nginx.conf -> http { ... }
+#
+# This defines the rate limit zone used by the-door's server block.
+# 10 requests per minute per IP, burst of 5 with nodelay.
+
+limit_req_zone $binary_remote_addr zone=the_door_api:10m rate=10r/m;
--- a/pytest.ini
+++ b/pytest.ini
@@ -0,0 +1,5 @@
+[pytest]
+testpaths = crisis
+python_files = tests.py
+python_classes = Test*
+python_functions = test_*