e678aa076b
Adds crisis detection and response system with 5-tier classification: - crisis/PROTOCOL.md: Crisis response protocol and tier definitions - crisis/detect.py: Tiered indicator engine (LOW/MEDIUM/HIGH/CRITICAL) - crisis/response.py: Timmy's crisis responses and UI flag generation - crisis/gateway.py: API gateway wrapper for crisis detection - crisis/tests.py: Unit tests for all crisis modules Integrates with existing crisis UI components in index.html. All smoke tests pass.
109 lines
3.1 KiB
Python
109 lines
3.1 KiB
Python
"""
|
|
Crisis Gateway Module for the-door.
|
|
|
|
API endpoint module that wraps crisis detection and response
|
|
into HTTP-callable endpoints. Integrates detect.py and response.py.
|
|
|
|
Usage:
|
|
from crisis.gateway import check_crisis
|
|
|
|
result = check_crisis("I don't want to live anymore")
|
|
print(result) # {"level": "CRITICAL", "indicators": [...], "response": {...}}
|
|
"""
|
|
|
|
import json
|
|
from typing import Optional
|
|
|
|
from .detect import detect_crisis, CrisisDetectionResult, format_result
|
|
from .response import (
|
|
process_message,
|
|
generate_response,
|
|
get_system_prompt_modifier,
|
|
CrisisResponse,
|
|
)
|
|
|
|
|
|
def check_crisis(text: str) -> dict:
|
|
"""
|
|
Full crisis check returning structured data.
|
|
|
|
Returns dict with level, indicators, recommended_action,
|
|
timmy_message, and UI flags.
|
|
"""
|
|
detection = detect_crisis(text)
|
|
response = generate_response(detection)
|
|
|
|
return {
|
|
"level": detection.level,
|
|
"score": detection.score,
|
|
"indicators": detection.indicators,
|
|
"recommended_action": detection.recommended_action,
|
|
"timmy_message": response.timmy_message,
|
|
"ui": {
|
|
"show_crisis_panel": response.show_crisis_panel,
|
|
"show_overlay": response.show_overlay,
|
|
"provide_988": response.provide_988,
|
|
},
|
|
"escalate": response.escalate,
|
|
}
|
|
|
|
|
|
def get_system_prompt(detection: CrisisDetectionResult) -> Optional[str]:
|
|
"""
|
|
Get the system prompt modifier for this detection level.
|
|
Returns None if no crisis detected.
|
|
"""
|
|
if detection.level == "NONE":
|
|
return None
|
|
return get_system_prompt_modifier(detection)
|
|
|
|
|
|
def format_gateway_response(text: str, pretty: bool = True) -> str:
|
|
"""
|
|
Full gateway response as formatted string or JSON.
|
|
|
|
This is the function that would be called by the gateway endpoint
|
|
when a message comes in.
|
|
"""
|
|
result = check_crisis(text)
|
|
|
|
if pretty:
|
|
return json.dumps(result, indent=2)
|
|
return json.dumps(result)
|
|
|
|
|
|
# ── Quick test interface ────────────────────────────────────────
|
|
|
|
def _interactive():
|
|
"""Interactive test mode."""
|
|
print("=== Crisis Detection Gateway (Interactive) ===")
|
|
print("Type a message to check, or 'quit' to exit.\n")
|
|
|
|
while True:
|
|
try:
|
|
user_input = input("You> ").strip()
|
|
except (EOFError, KeyboardInterrupt):
|
|
print("\nBye.")
|
|
break
|
|
|
|
if user_input.lower() in ("quit", "exit", "q"):
|
|
print("Bye.")
|
|
break
|
|
|
|
if not user_input:
|
|
continue
|
|
|
|
result = check_crisis(user_input)
|
|
print(f"\n Level: {result['level']}")
|
|
print(f" Score: {result['score']}")
|
|
print(f" Indicators: {', '.join(result['indicators']) if result['indicators'] else 'none'}")
|
|
print(f" Timmy says: {result['timmy_message']}")
|
|
print(f" Overlay: {result['ui']['show_overlay']}")
|
|
print(f" 988 banner: {result['ui']['provide_988']}")
|
|
print(f" Escalate: {result['escalate']}")
|
|
print()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
_interactive()
|