bb4ba82ac8
- 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
130 lines
3.7 KiB
Python
130 lines
3.7 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 .compassion_router import router
|
|
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(base_prompt: str, text: str = "") -> str:
|
|
"""
|
|
Sovereign Heart System Prompt Override.
|
|
|
|
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.
|
|
"""
|
|
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:
|
|
"""
|
|
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()
|