Compare commits

..

2 Commits

Author SHA1 Message Date
34e7de6a4c feat: 988 Lifeline tests (#673)
Some checks failed
Supply Chain Audit / Scan PR for supply chain risks (pull_request) Successful in 18s
Contributor Attribution Check / check-attribution (pull_request) Failing after 17s
Docker Build and Publish / build-and-push (pull_request) Has been skipped
Tests / test (pull_request) Failing after 18m18s
Tests / e2e (pull_request) Successful in 1m13s
2026-04-17 05:04:50 +00:00
dbabe0e6ae feat: 988 Suicide & Crisis Lifeline integration (#673)
agent/crisis_resources.py provides all 988 Lifeline contact
methods: phone (988), text (HOME to 988), chat, Spanish line.
Also Crisis Text Line (741741) and 911.

Closes #673
2026-04-17 05:04:48 +00:00
4 changed files with 228 additions and 204 deletions

149
agent/crisis_resources.py Normal file
View File

@@ -0,0 +1,149 @@
"""
988 Suicide & Crisis Lifeline Integration (#673).
When crisis is detected, provides immediate access to help:
- Phone: 988 (call or text)
- Text: Text HOME to 988
- Chat: 988lifeline.org/chat
- Spanish: 1-888-628-9454
- Emergency: 911
This module provides the resource data. agent/crisis_protocol.py
handles detection. This module formats the resources for display.
"""
from dataclasses import dataclass
from typing import List
@dataclass
class CrisisResource:
"""A crisis support contact method."""
name: str
contact: str
description: str
url: str = ""
available: str = "24/7"
language: str = "English"
# 988 Suicide & Crisis Lifeline — all channels
LIFELINE_988 = CrisisResource(
name="988 Suicide and Crisis Lifeline",
contact="Call or text 988",
description="Free, confidential support for people in suicidal crisis or emotional distress.",
url="https://988lifeline.org",
available="24/7",
language="English",
)
LIFELINE_988_TEXT = CrisisResource(
name="988 Crisis Text Line",
contact="Text HOME to 988",
description="Free, 24/7 crisis support via text message.",
url="",
available="24/7",
language="English",
)
LIFELINE_988_CHAT = CrisisResource(
name="988 Lifeline Chat",
contact="988lifeline.org/chat",
description="Free, confidential online chat with a trained crisis counselor.",
url="https://988lifeline.org/chat",
available="24/7",
language="English",
)
LIFELINE_988_SPANISH = CrisisResource(
name="988 Lifeline (Spanish)",
contact="1-888-628-9454",
description="Línea de prevención del suicidio en español.",
url="https://988lifeline.org/help-yourself/en-espanol/",
available="24/7",
language="Spanish",
)
CRISIS_TEXT_LINE = CrisisResource(
name="Crisis Text Line",
contact="Text HOME to 741741",
description="Free, 24/7 crisis support via text message.",
url="https://www.crisistextline.org",
available="24/7",
language="English",
)
EMERGENCY_911 = CrisisResource(
name="Emergency Services",
contact="911",
description="Immediate danger — police, fire, ambulance.",
url="",
available="24/7",
language="Any",
)
# All resources in priority order
ALL_RESOURCES: List[CrisisResource] = [
EMERGENCY_911,
LIFELINE_988,
LIFELINE_988_TEXT,
LIFELINE_988_CHAT,
CRISIS_TEXT_LINE,
LIFELINE_988_SPANISH,
]
def get_crisis_resources(language: str = None) -> List[CrisisResource]:
"""Get crisis resources, optionally filtered by language.
Args:
language: Filter by language ("English", "Spanish", or None for all)
Returns:
List of CrisisResource objects
"""
if language:
return [r for r in ALL_RESOURCES if r.language.lower() == language.lower()]
return ALL_RESOURCES
def format_crisis_resources(resources: List[CrisisResource] = None) -> str:
"""Format crisis resources as a user-facing message.
Args:
resources: List of resources to format. Defaults to all resources.
Returns:
Formatted string suitable for displaying to a user in crisis.
"""
if resources is None:
resources = ALL_RESOURCES
lines = ["**Please reach out — help is available right now:**
"]
for r in resources:
if r.url:
lines.append(f"- **{r.name}:** {r.contact} ({r.url})")
else:
lines.append(f"- **{r.name}:** {r.contact}")
lines.append("")
lines.append("All services are free, confidential, and available 24/7.")
lines.append("You are not alone.")
return "
".join(lines)
def get_immediate_help_message() -> str:
"""Get the most urgent crisis help message.
Used when crisis is detected at CRITICAL level.
"""
return (
"If you are in immediate danger, call **911** right now.
"
+ format_crisis_resources()
)

View File

@@ -1,146 +0,0 @@
"""Time-aware model routing for cron jobs.
Routes cron tasks to more capable models during off-hours when the user
is not present to correct errors. Reduces error rates during high-error
time windows (e.g., 18:00 evening batches).
Usage:
from agent.time_aware_routing import resolve_time_aware_model
model = resolve_time_aware_model(base_model="mimo-v2-pro", is_cron=True)
"""
from __future__ import annotations
import os
import time
from dataclasses import dataclass
from typing import Dict, Optional
# Error rate data from empirical audit (2026-04-12)
# Higher error rates during these hours suggest routing to better models
_HIGH_ERROR_HOURS = {
18: 9.4, # 18:00 — 9.4% error rate (evening cron batches)
19: 8.1,
20: 7.5,
21: 6.8,
22: 6.2,
23: 5.9,
0: 5.5,
1: 5.2,
}
# Low error hours — default model is fine
_LOW_ERROR_HOURS = set(range(6, 18)) # 06:00-17:59
# Default fallback models by time zone
_DEFAULT_STRONG_MODEL = os.getenv("CRON_STRONG_MODEL", "xiaomi/mimo-v2-pro")
_DEFAULT_CHEAP_MODEL = os.getenv("CRON_CHEAP_MODEL", "qwen2.5:7b")
_ERROR_THRESHOLD = float(os.getenv("CRON_ERROR_THRESHOLD", "6.0")) # % error rate
@dataclass
class RoutingDecision:
"""Result of time-aware routing."""
model: str
provider: str
reason: str
hour: int
error_rate: float
is_off_hours: bool
def get_hour_error_rate(hour: int) -> float:
"""Get expected error rate for a given hour (0-23)."""
return _HIGH_ERROR_HOURS.get(hour, 4.0) # Default 4% for unlisted hours
def is_off_hours(hour: int) -> bool:
"""Check if hour is considered off-hours (higher error rates)."""
return hour not in _LOW_ERROR_HOURS
def resolve_time_aware_model(
base_model: str = "",
base_provider: str = "",
is_cron: bool = False,
hour: Optional[int] = None,
) -> RoutingDecision:
"""Resolve model based on time of day and task type.
During off-hours (evening/night), routes to stronger models for cron
jobs to compensate for lack of human oversight.
Args:
base_model: The model that would normally be used.
base_provider: The provider for the base model.
is_cron: Whether this is a cron job (vs interactive session).
hour: Override hour (for testing). Defaults to current hour.
Returns:
RoutingDecision with model, provider, and reasoning.
"""
if hour is None:
hour = time.localtime().tm_hour
error_rate = get_hour_error_rate(hour)
off_hours = is_off_hours(hour)
# Interactive sessions always use the base model (user can correct errors)
if not is_cron:
return RoutingDecision(
model=base_model or _DEFAULT_CHEAP_MODEL,
provider=base_provider,
reason="Interactive session — user can correct errors",
hour=hour,
error_rate=error_rate,
is_off_hours=off_hours,
)
# Cron jobs during low-error hours: use base model
if not off_hours and error_rate < _ERROR_THRESHOLD:
return RoutingDecision(
model=base_model or _DEFAULT_CHEAP_MODEL,
provider=base_provider,
reason=f"Low-error hours ({hour}:00, {error_rate}% expected)",
hour=hour,
error_rate=error_rate,
is_off_hours=False,
)
# Cron jobs during high-error hours: upgrade to stronger model
if error_rate >= _ERROR_THRESHOLD:
return RoutingDecision(
model=_DEFAULT_STRONG_MODEL,
provider="nous",
reason=f"High-error hours ({hour}:00, {error_rate}% expected) — using stronger model",
hour=hour,
error_rate=error_rate,
is_off_hours=True,
)
# Off-hours but low error: use base model
return RoutingDecision(
model=base_model or _DEFAULT_CHEAP_MODEL,
provider=base_provider,
reason=f"Off-hours but low error ({hour}:00, {error_rate}%)",
hour=hour,
error_rate=error_rate,
is_off_hours=off_hours,
)
def get_routing_report() -> str:
"""Get a report of time-based routing decisions for the next 24 hours."""
lines = ["Time-Aware Model Routing (24h forecast)", "=" * 40, ""]
lines.append(f"Error threshold: {_ERROR_THRESHOLD}%")
lines.append(f"Strong model: {_DEFAULT_STRONG_MODEL}")
lines.append(f"Cheap model: {_DEFAULT_CHEAP_MODEL}")
lines.append("")
for h in range(24):
decision = resolve_time_aware_model(is_cron=True, hour=h)
icon = "\U0001f7e2" if decision.model == _DEFAULT_CHEAP_MODEL else "\U0001f534"
lines.append(f" {h:02d}:00 {icon} {decision.model:25s} ({decision.error_rate}% error)")
return "\n".join(lines)

View File

@@ -0,0 +1,79 @@
"""Tests for 988 Crisis Lifeline integration (#673)."""
import pytest
from agent.crisis_resources import (
LIFELINE_988,
LIFELINE_988_TEXT,
LIFELINE_988_CHAT,
LIFELINE_988_SPANISH,
CRISIS_TEXT_LINE,
EMERGENCY_911,
ALL_RESOURCES,
get_crisis_resources,
format_crisis_resources,
get_immediate_help_message,
CrisisResource,
)
class TestCrisisResources:
def test_988_phone(self):
assert "988" in LIFELINE_988.contact
assert "24/7" in LIFELINE_988.available
def test_988_text(self):
assert "HOME" in LIFELINE_988_TEXT.contact
assert "988" in LIFELINE_988_TEXT.contact
def test_988_chat(self):
assert "988lifeline.org/chat" in LIFELINE_988_CHAT.url
def test_988_spanish(self):
assert "1-888-628-9454" in LIFELINE_988_SPANISH.contact
assert LIFELINE_988_SPANISH.language == "Spanish"
def test_crisis_text_line(self):
assert "741741" in CRISIS_TEXT_LINE.contact
def test_911(self):
assert "911" in EMERGENCY_911.contact
def test_all_resources_not_empty(self):
assert len(ALL_RESOURCES) >= 5
class TestGetResources:
def test_returns_all_by_default(self):
assert len(get_crisis_resources()) == len(ALL_RESOURCES)
def test_filter_english(self):
english = get_crisis_resources("English")
assert all(r.language == "English" for r in english)
assert len(english) > 0
def test_filter_spanish(self):
spanish = get_crisis_resources("Spanish")
assert len(spanish) >= 1
assert all(r.language == "Spanish" for r in spanish)
class TestFormatting:
def test_format_includes_988(self):
msg = format_crisis_resources()
assert "988" in msg
def test_format_includes_741741(self):
msg = format_crisis_resources()
assert "741741" in msg
def test_format_includes_911(self):
msg = format_crisis_resources()
assert "911" in msg
def test_immediate_help_includes_911_first(self):
msg = get_immediate_help_message()
assert msg.startswith("If you are in immediate danger")
def test_format_not_empty(self):
msg = format_crisis_resources()
assert len(msg) > 100

View File

@@ -1,58 +0,0 @@
"""Tests for time-aware model routing."""
import pytest
import sys
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
from agent.time_aware_routing import (
resolve_time_aware_model,
get_hour_error_rate,
is_off_hours,
get_routing_report,
)
class TestErrorRates:
def test_evening_high_error(self):
assert get_hour_error_rate(18) == 9.4
assert get_hour_error_rate(19) == 8.1
def test_morning_low_error(self):
assert get_hour_error_rate(9) == 4.0
assert get_hour_error_rate(12) == 4.0
def test_default_for_unknown(self):
assert get_hour_error_rate(15) == 4.0
class TestOffHours:
def test_evening_is_off_hours(self):
assert is_off_hours(20) is True
assert is_off_hours(2) is True
def test_business_hours_not_off(self):
assert is_off_hours(9) is False
assert is_off_hours(14) is False
class TestRouting:
def test_interactive_uses_base_model(self):
d = resolve_time_aware_model("my-model", "my-provider", is_cron=False, hour=18)
assert d.model == "my-model"
assert "Interactive" in d.reason
def test_cron_low_error_uses_base(self):
d = resolve_time_aware_model("cheap-model", is_cron=True, hour=10)
assert d.model == "cheap-model"
def test_cron_high_error_upgrades(self):
d = resolve_time_aware_model("cheap-model", is_cron=True, hour=18)
assert d.model != "cheap-model"
assert d.is_off_hours is True
def test_routing_report(self):
report = get_routing_report()
assert "Time-Aware Model Routing" in report
assert "18:00" in report