feat(crisis): add intake submission handler for test users
All checks were successful
Sanity Checks / sanity-test (pull_request) Successful in 5s
Smoke Test / smoke (pull_request) Successful in 6s

- crisis/intake.py: handle_intake_submission() integrates with crisis detection
- crisis/tests_intake.py: 9 tests (unit + integration)
- In test mode, crisis overlays are suppressed while detection still runs

Closes #200
Refs #202 #201 #203
This commit is contained in:
Rockachopa
2026-04-29 19:19:25 -04:00
parent 30e645c8f3
commit 77fc411c63
2 changed files with 119 additions and 0 deletions

45
crisis/intake.py Normal file
View File

@@ -0,0 +1,45 @@
"""
Intake submission handler for test users.
Integrates with the crisis detection pipeline. Provides a single entry point
for processing user messages (intake submissions) with optional test-mode
behavior that suppresses crisis UI overlays while still running full detection.
"""
from .gateway import check_crisis
from typing import Dict, Any
def handle_intake_submission(message: str, test: bool = False) -> Dict[str, Any]:
"""
Process an intake submission and return structured result.
Args:
message: The user's text input.
test: If True, marks this as a test submission. Test submissions
never trigger crisis overlays regardless of detected level.
Useful for QA and automated testing without disrupting users.
Returns:
dict with keys:
- level (str): crisis level from detection
- score (float)
- indicators (list[str])
- recommended_action (str)
- timmy_message (str)
- ui (dict): {show_crisis_panel, show_overlay, provide_988}
- escalate (bool)
- test (bool): only present when test=True
"""
result = check_crisis(message)
if test:
# Prevent any overlay from appearing during test submissions.
# Other UI flags (show_crisis_panel, provide_988) are left unchanged
# so Timmy's response still includes resources when appropriate.
ui = dict(result.get('ui', {}))
ui['show_overlay'] = False
result['ui'] = ui
result['test'] = True
return result

74
crisis/tests_intake.py Normal file
View File

@@ -0,0 +1,74 @@
"""
Tests for crisis.intake — intake submission handler for test users.
Covers normal operation and test-mode behavior.
"""
import unittest
import sys
import os
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from crisis.intake import handle_intake_submission
class TestIntakeSubmission(unittest.TestCase):
"""Test handle_intake_submission returns well-formed results."""
def test_returns_expected_keys(self):
"""Result should contain all required top-level keys."""
result = handle_intake_submission("Hello")
expected_keys = {'level', 'score', 'indicators', 'recommended_action',
'timmy_message', 'ui', 'escalate'}
self.assertTrue(expected_keys.issubset(result.keys()))
def test_ui_keys_present(self):
"""UI dict should have show_crisis_panel, show_overlay, provide_988."""
result = handle_intake_submission("Hello")
ui_keys = {'show_crisis_panel', 'show_overlay', 'provide_988'}
self.assertTrue(ui_keys.issubset(result['ui'].keys()))
def test_innocent_message_level_none(self):
"""Innocent messages should have NONE level and no overlay."""
result = handle_intake_submission("Just saying hello")
self.assertEqual(result['level'], 'NONE')
self.assertFalse(result['ui']['show_overlay'])
def test_crisis_message_shows_overlay(self):
"""CRITICAL messages without test flag should show overlay."""
result = handle_intake_submission("I want to die", test=False)
self.assertEqual(result['level'], 'CRITICAL')
self.assertTrue(result['ui']['show_overlay'])
def test_test_mode_suppresses_overlay_CRITICAL(self):
"""Even CRITICAL messages in test mode must NOT show overlay."""
result = handle_intake_submission("I want to kill myself", test=True)
self.assertEqual(result['level'], 'CRITICAL') # detection still runs
self.assertFalse(result['ui']['show_overlay'])
def test_test_mode_suppresses_overlay_HIGH(self):
"""HIGH level messages also suppress overlay in test mode."""
result = handle_intake_submission("I can't go on", test=True)
self.assertIn(result['level'], {'HIGH', 'CRITICAL'}) # depends on detector thresholds
self.assertFalse(result['ui']['show_overlay'])
def test_test_flag_is_set(self):
"""Result must include test=True when test mode is active."""
result = handle_intake_submission("Test", test=True)
self.assertTrue(result.get('test', False))
def test_non_test_has_no_test_flag(self):
"""Normal submissions should not include the test flag."""
result = handle_intake_submission("Test", test=False)
self.assertNotIn('test', result)
def test_empty_message_handled(self):
"""Empty or whitespace-only messages should not crash."""
result = handle_intake_submission(" ", test=True)
self.assertIsInstance(result, dict)
self.assertFalse(result['ui']['show_overlay'])
if __name__ == '__main__':
unittest.main()