From 77fc411c63ac2574f64116f0b2601ee741bb9519 Mon Sep 17 00:00:00 2001 From: Rockachopa Date: Wed, 29 Apr 2026 19:19:25 -0400 Subject: [PATCH] feat(crisis): add intake submission handler for test users - 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 --- crisis/intake.py | 45 +++++++++++++++++++++++++ crisis/tests_intake.py | 74 ++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 119 insertions(+) create mode 100644 crisis/intake.py create mode 100644 crisis/tests_intake.py diff --git a/crisis/intake.py b/crisis/intake.py new file mode 100644 index 0000000..e020808 --- /dev/null +++ b/crisis/intake.py @@ -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 diff --git a/crisis/tests_intake.py b/crisis/tests_intake.py new file mode 100644 index 0000000..6599e85 --- /dev/null +++ b/crisis/tests_intake.py @@ -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() -- 2.43.0