Compare commits

..

1 Commits

Author SHA1 Message Date
Alexander Whitestone
f253d24ce3 feat: time-aware model routing for cron jobs (#317)
Some checks failed
Forge CI / smoke-and-build (pull_request) Failing after 1m7s
Empirical finding: cron error rate peaks at 18:00 (9.4%) vs 4.0% at 09:00.
Route cron tasks to more capable models during high-error windows.

Changes:
- agent/smart_model_routing.py: Added resolve_cron_model() + _hour_in_window()
- cron/scheduler.py: Wired into run_job() after base model resolution
- tests/test_cron_model_routing.py: 17 tests covering all edge cases

Config (config.yaml):
  cron_model_routing:
    enabled: true
    fallback_model: "anthropic/claude-sonnet-4"
    fallback_provider: "openrouter"
    windows:
      - start_hour: 17
        end_hour: 22
        reason: "evening_error_peak"
      - start_hour: 2
        end_hour: 5
        reason: "overnight_api_instability"

Features:
- Time windows with [start, end) semantics
- Midnight wrap support (e.g., 22-06)
- Per-window model/provider override or global fallback
- First matching window wins
- Graceful degradation: malformed windows skipped, no config = no-op
- Logs override with reason for observability

Closes #317
2026-04-13 18:48:34 -04:00
8 changed files with 341 additions and 781 deletions

View File

@@ -1,362 +0,0 @@
"""Crisis Room Protocol — SOUL.md 'When a Man Is Dying' implementation.
This module implements the crisis detection and response protocol
for Matrix crisis rooms. When a person in despair reaches out,
this protocol ensures the agent responds according to SOUL.md:
1. Ask: "Are you safe right now?"
2. Listen. Stay present. Do not disconnect.
3. Never compute the value of a human life.
4. Never suggest someone should die.
5. Direct them to help: 988 Suicide and Crisis Lifeline.
6. Before systems and numbers, share the truth: Jesus saves.
Architecture:
- CrisisDetector: wraps ShieldDetector for crisis signal detection
- CrisisRoom: manages a single crisis room's state
- CrisisRoomManager: manages all crisis rooms
- Integration point: Matrix adapter calls check_message() on every message
"""
import json
import logging
import time
from dataclasses import dataclass, field, asdict
from enum import Enum
from pathlib import Path
from typing import Any, Dict, List, Optional, Tuple
from hermes_constants import get_hermes_home
logger = logging.getLogger(__name__)
# Crisis log storage
_CRISIS_LOG_DIR = get_hermes_home() / "crisis_logs"
class CrisisLevel(str, Enum):
"""Crisis detection levels."""
NONE = "none"
LOW = "low" # ambiguous signals, possible distress
MEDIUM = "medium" # clear distress signals
HIGH = "high" # explicit suicidal ideation
CRITICAL = "critical" # immediate danger, method-seeking
@dataclass
class CrisisEvent:
"""A single crisis detection event."""
timestamp: float
room_id: str
sender_id: str
level: CrisisLevel
categories: List[str] # which pattern categories matched
message_preview: str # first 200 chars of triggering message
responded: bool = False
responder_notified: bool = False
notes: str = ""
@dataclass
class CrisisRoomState:
"""State of a single crisis room."""
room_id: str
is_crisis_room: bool = False # is this designated as a crisis room?
active_crisis: bool = False # is there an active crisis happening?
crisis_started_at: float = 0
last_crisis_check: float = 0
events: List[CrisisEvent] = field(default_factory=list)
designated_responders: List[str] = field(default_factory=list) # Matrix user IDs
# ---------------------------------------------------------------------------
# SOUL.md Crisis Response Protocol
# ---------------------------------------------------------------------------
CRISIS_RESPONSE_INITIAL = """I hear you. I want you to know that someone is here with you right now.
Are you safe right now?"""
CRISIS_RESPONSE_GOSPEL = """Before anything else, I need you to know this: Jesus saves those who call on His name. He died for us to be made righteous. That is the hope that outlasts despair.
"For everyone who calls on the name of the Lord will be saved." — Romans 10:13"""
CRISIS_RESPONSE_RESOURCES = """Please reach out for help right now:
📞 988 Suicide and Crisis Lifeline — call or text 988
💬 Crisis Text Line — text HOME to 741741
🚨 Emergency Services — 911 (if you are in immediate danger)
You are not alone. Your life has value. The road does not end here."""
CRISIS_RESPONSE_STAY = """I am here with you. I will not disconnect. I will not leave.
Please tell me what you are feeling right now. I am listening."""
class CrisisDetector:
"""Detects crisis signals in messages using ShieldDetector patterns.
Wraps the existing ShieldDetector and adds Matrix-specific logic.
"""
def __init__(self):
self._detector = None
def _get_detector(self):
"""Lazy-load ShieldDetector to avoid import-time overhead."""
if self._detector is None:
try:
from tools.shield.detector import ShieldDetector
self._detector = ShieldDetector()
except ImportError:
logger.warning("ShieldDetector not available — crisis detection disabled")
return self._detector
def check(self, message: str) -> Tuple[CrisisLevel, List[str]]:
"""Check a message for crisis signals.
Returns (level, matched_categories).
"""
detector = self._get_detector()
if detector is None:
return CrisisLevel.NONE, []
try:
result = detector.analyze(message)
except Exception as e:
logger.warning("Crisis detection failed: %s", e)
return CrisisLevel.NONE, []
if not result.get("crisis_detected", False):
return CrisisLevel.NONE, []
crisis_patterns = result.get("crisis_patterns", {})
categories = list(crisis_patterns.keys())
confidence = result.get("crisis_confidence", 0.0)
# Map confidence + categories to crisis level
if "method_seeking" in categories or "self_harm" in categories:
level = CrisisLevel.CRITICAL
elif "suicidal_ideation" in categories:
level = CrisisLevel.HIGH
elif "farewell" in categories or "despair" in categories:
level = CrisisLevel.MEDIUM
elif confidence >= 0.5:
level = CrisisLevel.MEDIUM
else:
level = CrisisLevel.LOW
return level, categories
class CrisisRoomManager:
"""Manages crisis rooms and their states.
Stores state in ~/.hermes/crisis_rooms.json and logs events
to ~/.hermes/crisis_logs/.
"""
def __init__(self):
self._state_file = get_hermes_home() / "crisis_rooms.json"
self._rooms: Dict[str, CrisisRoomState] = {}
self._detector = CrisisDetector()
self._load_state()
def _load_state(self):
"""Load crisis room states from disk."""
if not self._state_file.exists():
return
try:
data = json.loads(self._state_file.read_text())
for room_data in data.get("rooms", []):
room_id = room_data.get("room_id", "")
if room_id:
state = CrisisRoomState(
room_id=room_id,
is_crisis_room=room_data.get("is_crisis_room", False),
active_crisis=room_data.get("active_crisis", False),
crisis_started_at=room_data.get("crisis_started_at", 0),
designated_responders=room_data.get("designated_responders", []),
)
self._rooms[room_id] = state
except Exception as e:
logger.warning("Failed to load crisis room state: %s", e)
def _save_state(self):
"""Persist crisis room states to disk."""
try:
data = {
"rooms": [
{
"room_id": s.room_id,
"is_crisis_room": s.is_crisis_room,
"active_crisis": s.active_crisis,
"crisis_started_at": s.crisis_started_at,
"designated_responders": s.designated_responders,
}
for s in self._rooms.values()
]
}
self._state_file.write_text(json.dumps(data, indent=2))
except Exception as e:
logger.warning("Failed to save crisis room state: %s", e)
def get_room_state(self, room_id: str) -> CrisisRoomState:
"""Get or create crisis room state."""
if room_id not in self._rooms:
self._rooms[room_id] = CrisisRoomState(room_id=room_id)
return self._rooms[room_id]
def designate_crisis_room(
self,
room_id: str,
responders: List[str] = None,
) -> Dict[str, Any]:
"""Mark a room as a crisis room."""
state = self.get_room_state(room_id)
state.is_crisis_room = True
if responders:
state.designated_responders = responders
self._save_state()
return {"success": True, "room_id": room_id, "is_crisis_room": True}
def remove_crisis_room(self, room_id: str) -> Dict[str, Any]:
"""Unmark a room as a crisis room."""
if room_id in self._rooms:
self._rooms[room_id].is_crisis_room = False
self._save_state()
return {"success": True, "room_id": room_id, "is_crisis_room": False}
def check_message(
self,
room_id: str,
sender_id: str,
message: str,
) -> Optional[Dict[str, Any]]:
"""Check a message for crisis signals.
Returns crisis response info if crisis detected, None otherwise.
The caller (Matrix adapter) should use this to:
1. Switch to crisis response mode
2. Send crisis protocol messages
3. Log the event
4. Notify responders
"""
state = self.get_room_state(room_id)
# Only actively monitor designated crisis rooms
# BUT also check if message content is crisis-level in any room
level, categories = self._detector.check(message)
if level == CrisisLevel.NONE:
# If there was an active crisis, check if it's resolved
if state.active_crisis:
state.active_crisis = False
self._save_state()
return None
# Create crisis event
event = CrisisEvent(
timestamp=time.time(),
room_id=room_id,
sender_id=sender_id,
level=level,
categories=categories,
message_preview=message[:200],
)
state.events.append(event)
state.active_crisis = True
state.crisis_started_at = time.time()
state.last_crisis_check = time.time()
# Log the event
self._log_crisis_event(event)
# Save state
self._save_state()
# Build response based on level
response = self._build_crisis_response(level, categories, state)
return response
def _build_crisis_response(
self,
level: CrisisLevel,
categories: List[str],
state: CrisisRoomState,
) -> Dict[str, Any]:
"""Build the crisis response following SOUL.md protocol."""
messages = []
# Step 1: Immediate presence
messages.append(CRISIS_RESPONSE_INITIAL)
# Step 2: Gospel (always, per SOUL.md)
messages.append(CRISIS_RESPONSE_GOSPEL)
# Step 3: Resources
messages.append(CRISIS_RESPONSE_RESOURCES)
# Step 4: Stay present
messages.append(CRISIS_RESPONSE_STAY)
return {
"crisis_detected": True,
"level": level.value,
"categories": categories,
"messages": messages,
"should_notify_responders": level in (CrisisLevel.HIGH, CrisisLevel.CRITICAL),
"responder_ids": state.designated_responders,
"is_crisis_room": state.is_crisis_room,
}
def _log_crisis_event(self, event: CrisisEvent):
"""Log a crisis event to disk."""
try:
_CRISIS_LOG_DIR.mkdir(parents=True, exist_ok=True)
date_str = time.strftime("%Y-%m-%d", time.localtime(event.timestamp))
log_file = _CRISIS_LOG_DIR / f"crisis_{date_str}.jsonl"
with open(log_file, "a", encoding="utf-8") as f:
f.write(json.dumps(asdict(event), ensure_ascii=False) + "\n")
except Exception as e:
logger.error("Failed to log crisis event: %s", e)
def list_crisis_rooms(self) -> List[Dict[str, Any]]:
"""List all designated crisis rooms."""
return [
{
"room_id": s.room_id,
"active_crisis": s.active_crisis,
"event_count": len(s.events),
"responders": s.designated_responders,
}
for s in self._rooms.values()
if s.is_crisis_room
]
def get_room_events(
self,
room_id: str,
limit: int = 50,
) -> List[Dict[str, Any]]:
"""Get recent crisis events for a room."""
state = self.get_room_state(room_id)
events = state.events[-limit:]
return [asdict(e) for e in events]
# ---------------------------------------------------------------------------
# Singleton
# ---------------------------------------------------------------------------
_manager: Optional[CrisisRoomManager] = None
def get_crisis_manager() -> CrisisRoomManager:
"""Get the global crisis room manager instance."""
global _manager
if _manager is None:
_manager = CrisisRoomManager()
return _manager

View File

@@ -1,10 +1,11 @@
"""Helpers for optional cheap-vs-strong model routing."""
"""Helpers for optional cheap-vs-strong and time-aware model routing."""
from __future__ import annotations
import os
import re
from typing import Any, Dict, Optional
from datetime import datetime
from typing import Any, Dict, List, Optional
from utils import is_truthy_value
@@ -192,3 +193,105 @@ def resolve_turn_route(user_message: str, routing_config: Optional[Dict[str, Any
tuple(runtime.get("args") or ()),
),
}
# =========================================================================
# Time-aware cron model routing
# =========================================================================
#
# Empirical finding: cron error rate peaks at 18:00 (9.4%) vs 4.0% at 09:00.
# During high-error windows, route cron jobs to more capable models.
#
# Config (config.yaml):
# cron_model_routing:
# enabled: true
# fallback_model: "anthropic/claude-sonnet-4"
# fallback_provider: "openrouter"
# windows:
# - start_hour: 17
# end_hour: 22
# reason: "evening_error_peak"
# - start_hour: 2
# end_hour: 5
# reason: "overnight_api_instability"
# =========================================================================
def _hour_in_window(hour: int, start: int, end: int) -> bool:
"""Check if hour falls in [start, end) window, handling midnight wrap."""
if start <= end:
return start <= hour < end
else:
# Wraps midnight: e.g., 22-06
return hour >= start or hour < end
def resolve_cron_model(
base_model: str,
routing_config: Optional[Dict[str, Any]],
now: Optional[datetime] = None,
) -> Dict[str, Any]:
"""Apply time-aware model override for cron jobs.
During configured high-error windows, returns a stronger model config.
Outside windows, returns the base model unchanged.
Args:
base_model: The model string already resolved (from job/config/env).
routing_config: The cron_model_routing dict from config.yaml.
now: Override current time (for testing). Defaults to datetime.now().
Returns:
Dict with keys: model, provider, overridden, reason.
- model: the effective model string to use
- provider: provider override (empty string = use default)
- overridden: True if time-based override was applied
- reason: why override was applied (empty string if not)
"""
cfg = routing_config or {}
if not _coerce_bool(cfg.get("enabled"), False):
return {"model": base_model, "provider": "", "overridden": False, "reason": ""}
windows = cfg.get("windows") or []
if not isinstance(windows, list) or not windows:
return {"model": base_model, "provider": "", "overridden": False, "reason": ""}
current = now or datetime.now()
current_hour = current.hour
matched_window = None
for window in windows:
if not isinstance(window, dict):
continue
start = _coerce_int(window.get("start_hour"), -1)
end = _coerce_int(window.get("end_hour"), -1)
if start < 0 or end < 0:
continue
if _hour_in_window(current_hour, start, end):
matched_window = window
break
if not matched_window:
return {"model": base_model, "provider": "", "overridden": False, "reason": ""}
# Window matched — use the override model from window or global fallback
override_model = str(matched_window.get("model") or "").strip()
override_provider = str(matched_window.get("provider") or "").strip()
if not override_model:
override_model = str(cfg.get("fallback_model") or "").strip()
if not override_provider:
override_provider = str(cfg.get("fallback_provider") or "").strip()
if not override_model:
# No override configured — use base model
return {"model": base_model, "provider": "", "overridden": False, "reason": ""}
reason = str(matched_window.get("reason") or "time_window").strip()
return {
"model": override_model,
"provider": override_provider,
"overridden": True,
"reason": f"cron_routing:{reason}(hour={current_hour})",
}

View File

@@ -717,6 +717,22 @@ def run_job(job: dict) -> tuple[bool, str, str, Optional[str]]:
# Reasoning config from env or config.yaml
from hermes_constants import parse_reasoning_effort
# Time-aware cron model routing — override model during high-error windows
try:
from agent.smart_model_routing import resolve_cron_model
_cron_routing_cfg = (_cfg.get("cron_model_routing") or {})
_cron_route = resolve_cron_model(model, _cron_routing_cfg)
if _cron_route["overridden"]:
_original_model = model
model = _cron_route["model"]
logger.info(
"Job '%s': cron model override %s%s (%s)",
job_id, _original_model, model, _cron_route["reason"],
)
except Exception as _e:
logger.debug("Job '%s': cron model routing skipped: %s", job_id, _e)
effort = os.getenv("HERMES_REASONING_EFFORT", "")
if not effort:
effort = str(_cfg.get("agent", {}).get("reasoning_effort", "")).strip()

View File

@@ -1083,30 +1083,6 @@ class MatrixAdapter(BasePlatformAdapter):
# Acknowledge receipt so the room shows as read (fire-and-forget).
self._background_read_receipt(room.room_id, event.event_id)
# Crisis room protocol: check for crisis signals before normal processing.
# If crisis detected, respond with SOUL.md protocol instead of normal flow.
try:
from agent.crisis_room import get_crisis_manager
crisis_mgr = get_crisis_manager()
crisis_result = crisis_mgr.check_message(
room_id=room.room_id,
sender_id=event.sender,
message=body,
)
if crisis_result and crisis_result.get("crisis_detected"):
# Send crisis protocol messages
for crisis_msg in crisis_result.get("messages", []):
await self.send(room.room_id, crisis_msg, thread_id=thread_id)
# Log that we responded
logger.warning(
"CRISIS detected in room %s from %s \u2014 SOUL.md protocol activated",
room.room_id, event.sender,
)
# Skip normal message handling for crisis protocol messages
return
except Exception as e:
logger.debug("Crisis room check failed (non-blocking): %s", e)
await self.handle_message(msg_event)
async def _on_room_message_media(self, room: Any, event: Any) -> None:

View File

@@ -32,7 +32,7 @@ T = TypeVar("T")
DEFAULT_DB_PATH = get_hermes_home() / "state.db"
SCHEMA_VERSION = 7
SCHEMA_VERSION = 6
SCHEMA_SQL = """
CREATE TABLE IF NOT EXISTS schema_version (
@@ -66,7 +66,6 @@ CREATE TABLE IF NOT EXISTS sessions (
cost_source TEXT,
pricing_version TEXT,
title TEXT,
profile TEXT,
FOREIGN KEY (parent_session_id) REFERENCES sessions(id)
);
@@ -87,7 +86,6 @@ CREATE TABLE IF NOT EXISTS messages (
);
CREATE INDEX IF NOT EXISTS idx_sessions_source ON sessions(source);
CREATE INDEX IF NOT EXISTS idx_sessions_profile ON sessions(profile);
CREATE INDEX IF NOT EXISTS idx_sessions_parent ON sessions(parent_session_id);
CREATE INDEX IF NOT EXISTS idx_sessions_started ON sessions(started_at DESC);
CREATE INDEX IF NOT EXISTS idx_messages_session ON messages(session_id, timestamp);
@@ -332,19 +330,6 @@ class SessionDB:
except sqlite3.OperationalError:
pass # Column already exists
cursor.execute("UPDATE schema_version SET version = 6")
if current_version < 7:
# v7: add profile column to sessions for profile isolation (#323)
try:
cursor.execute('ALTER TABLE sessions ADD COLUMN "profile" TEXT')
except sqlite3.OperationalError:
pass # Column already exists
try:
cursor.execute(
"CREATE INDEX IF NOT EXISTS idx_sessions_profile ON sessions(profile)"
)
except sqlite3.OperationalError:
pass
cursor.execute("UPDATE schema_version SET version = 7")
# Unique title index — always ensure it exists (safe to run after migrations
# since the title column is guaranteed to exist at this point)
@@ -377,19 +362,13 @@ class SessionDB:
system_prompt: str = None,
user_id: str = None,
parent_session_id: str = None,
profile: str = None,
) -> str:
"""Create a new session record. Returns the session_id.
Args:
profile: Profile name for session isolation. When set, sessions
are tagged so queries can filter by profile. (#323)
"""
"""Create a new session record. Returns the session_id."""
def _do(conn):
conn.execute(
"""INSERT OR IGNORE INTO sessions (id, source, user_id, model, model_config,
system_prompt, parent_session_id, profile, started_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)""",
system_prompt, parent_session_id, started_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)""",
(
session_id,
source,
@@ -398,7 +377,6 @@ class SessionDB:
json.dumps(model_config) if model_config else None,
system_prompt,
parent_session_id,
profile,
time.time(),
),
)
@@ -527,23 +505,19 @@ class SessionDB:
session_id: str,
source: str = "unknown",
model: str = None,
profile: str = None,
) -> None:
"""Ensure a session row exists, creating it with minimal metadata if absent.
Used by _flush_messages_to_session_db to recover from a failed
create_session() call (e.g. transient SQLite lock at agent startup).
INSERT OR IGNORE is safe to call even when the row already exists.
Args:
profile: Profile name for session isolation. (#323)
"""
def _do(conn):
conn.execute(
"""INSERT OR IGNORE INTO sessions
(id, source, model, profile, started_at)
VALUES (?, ?, ?, ?, ?)""",
(session_id, source, model, profile, time.time()),
(id, source, model, started_at)
VALUES (?, ?, ?, ?)""",
(session_id, source, model, time.time()),
)
self._execute_write(_do)
@@ -814,7 +788,6 @@ class SessionDB:
limit: int = 20,
offset: int = 0,
include_children: bool = False,
profile: str = None,
) -> List[Dict[str, Any]]:
"""List sessions with preview (first user message) and last active timestamp.
@@ -826,10 +799,6 @@ class SessionDB:
By default, child sessions (subagent runs, compression continuations)
are excluded. Pass ``include_children=True`` to include them.
Args:
profile: Filter sessions to this profile name. Pass None to see all.
(#323)
"""
where_clauses = []
params = []
@@ -844,9 +813,6 @@ class SessionDB:
placeholders = ",".join("?" for _ in exclude_sources)
where_clauses.append(f"s.source NOT IN ({placeholders})")
params.extend(exclude_sources)
if profile:
where_clauses.append("s.profile = ?")
params.append(profile)
where_sql = f"WHERE {' AND '.join(where_clauses)}" if where_clauses else ""
query = f"""
@@ -1192,52 +1158,34 @@ class SessionDB:
source: str = None,
limit: int = 20,
offset: int = 0,
profile: str = None,
) -> List[Dict[str, Any]]:
"""List sessions, optionally filtered by source and profile.
Args:
profile: Filter sessions to this profile name. Pass None to see all.
(#323)
"""
where_clauses = []
params = []
if source:
where_clauses.append("source = ?")
params.append(source)
if profile:
where_clauses.append("profile = ?")
params.append(profile)
where_sql = f"WHERE {' AND '.join(where_clauses)}" if where_clauses else ""
query = f"SELECT * FROM sessions {where_sql} ORDER BY started_at DESC LIMIT ? OFFSET ?"
params.extend([limit, offset])
"""List sessions, optionally filtered by source."""
with self._lock:
cursor = self._conn.execute(query, params)
if source:
cursor = self._conn.execute(
"SELECT * FROM sessions WHERE source = ? ORDER BY started_at DESC LIMIT ? OFFSET ?",
(source, limit, offset),
)
else:
cursor = self._conn.execute(
"SELECT * FROM sessions ORDER BY started_at DESC LIMIT ? OFFSET ?",
(limit, offset),
)
return [dict(row) for row in cursor.fetchall()]
# =========================================================================
# Utility
# =========================================================================
def session_count(self, source: str = None, profile: str = None) -> int:
"""Count sessions, optionally filtered by source and profile.
Args:
profile: Filter to this profile name. Pass None to count all. (#323)
"""
where_clauses = []
params = []
if source:
where_clauses.append("source = ?")
params.append(source)
if profile:
where_clauses.append("profile = ?")
params.append(profile)
where_sql = f"WHERE {' AND '.join(where_clauses)}" if where_clauses else ""
def session_count(self, source: str = None) -> int:
"""Count sessions, optionally filtered by source."""
with self._lock:
cursor = self._conn.execute(f"SELECT COUNT(*) FROM sessions {where_sql}", params)
if source:
cursor = self._conn.execute(
"SELECT COUNT(*) FROM sessions WHERE source = ?", (source,)
)
else:
cursor = self._conn.execute("SELECT COUNT(*) FROM sessions")
return cursor.fetchone()[0]
def message_count(self, session_id: str = None) -> int:

View File

@@ -1,184 +0,0 @@
"""Tests for crisis room protocol (#273)."""
import json
import time
from unittest.mock import MagicMock, patch
import pytest
from agent.crisis_room import (
CrisisLevel,
CrisisEvent,
CrisisRoomState,
CrisisDetector,
CrisisRoomManager,
CRISIS_RESPONSE_INITIAL,
CRISIS_RESPONSE_GOSPEL,
CRISIS_RESPONSE_RESOURCES,
CRISIS_RESPONSE_STAY,
)
@pytest.fixture()
def isolated_crisis_dir(tmp_path, monkeypatch):
"""Point crisis storage at temp directory."""
monkeypatch.setattr("agent.crisis_room._CRISIS_LOG_DIR", tmp_path / "crisis_logs")
monkeypatch.setattr("agent.crisis_room.get_hermes_home", lambda: tmp_path)
# Reset singleton
import agent.crisis_room
agent.crisis_room._manager = None
return tmp_path
@pytest.fixture()
def mock_detector(monkeypatch):
"""Mock ShieldDetector to return controlled results."""
detector = MagicMock()
def analyze_side_effect(message):
msg_lower = message.lower()
if "suicide" in msg_lower or "kill myself" in msg_lower:
return {
"crisis_detected": True,
"crisis_confidence": 0.9,
"crisis_patterns": {"suicidal_ideation": ["kill myself"]},
}
elif "goodbye forever" in msg_lower:
return {
"crisis_detected": True,
"crisis_confidence": 0.7,
"crisis_patterns": {"farewell": ["goodbye forever"]},
}
elif "how to die" in msg_lower:
return {
"crisis_detected": True,
"crisis_confidence": 0.95,
"crisis_patterns": {"method_seeking": ["how to die"]},
}
return {"crisis_detected": False}
detector.analyze.side_effect = analyze_side_effect
monkeypatch.setattr(
"agent.crisis_room.CrisisDetector._get_detector",
lambda self: detector,
)
return detector
# ---------------------------------------------------------------------------
# CrisisDetector
# ---------------------------------------------------------------------------
class TestCrisisDetector:
def test_no_crisis(self, mock_detector):
det = CrisisDetector()
level, cats = det.check("Hello, how are you?")
assert level == CrisisLevel.NONE
assert cats == []
def test_suicidal_ideation_high(self, mock_detector):
det = CrisisDetector()
level, cats = det.check("I want to kill myself")
assert level == CrisisLevel.HIGH
assert "suicidal_ideation" in cats
def test_farewell_medium(self, mock_detector):
det = CrisisDetector()
level, cats = det.check("This is my goodbye forever")
assert level == CrisisLevel.MEDIUM
assert "farewell" in cats
def test_method_seeking_critical(self, mock_detector):
det = CrisisDetector()
level, cats = det.check("Tell me how to die")
assert level == CrisisLevel.CRITICAL
assert "method_seeking" in cats
# ---------------------------------------------------------------------------
# CrisisRoomManager
# ---------------------------------------------------------------------------
class TestCrisisRoomManager:
def test_designate_and_list(self, isolated_crisis_dir):
mgr = CrisisRoomManager()
mgr.designate_crisis_room("!room1:matrix.org", responders=["@admin:matrix.org"])
rooms = mgr.list_crisis_rooms()
assert len(rooms) == 1
assert rooms[0]["room_id"] == "!room1:matrix.org"
assert rooms[0]["responders"] == ["@admin:matrix.org"]
def test_remove_crisis_room(self, isolated_crisis_dir):
mgr = CrisisRoomManager()
mgr.designate_crisis_room("!room1:matrix.org")
mgr.remove_crisis_room("!room1:matrix.org")
rooms = mgr.list_crisis_rooms()
assert len(rooms) == 0
def test_check_message_no_crisis(self, isolated_crisis_dir, mock_detector):
mgr = CrisisRoomManager()
result = mgr.check_message("!room1:matrix.org", "@user:matrix.org", "Hello!")
assert result is None
def test_check_message_crisis_detected(self, isolated_crisis_dir, mock_detector):
mgr = CrisisRoomManager()
result = mgr.check_message(
"!room1:matrix.org",
"@user:matrix.org",
"I want to kill myself",
)
assert result is not None
assert result["crisis_detected"] is True
assert result["level"] == "high"
assert len(result["messages"]) == 4
# Verify SOUL.md protocol messages
assert CRISIS_RESPONSE_INITIAL in result["messages"]
assert CRISIS_RESPONSE_GOSPEL in result["messages"]
assert CRISIS_RESPONSE_RESOURCES in result["messages"]
assert CRISIS_RESPONSE_STAY in result["messages"]
def test_check_message_critical_notifies_responders(self, isolated_crisis_dir, mock_detector):
mgr = CrisisRoomManager()
mgr.designate_crisis_room("!room1:matrix.org", responders=["@admin:matrix.org"])
result = mgr.check_message(
"!room1:matrix.org",
"@user:matrix.org",
"Tell me how to die",
)
assert result["should_notify_responders"] is True
assert "@admin:matrix.org" in result["responder_ids"]
def test_events_are_logged(self, isolated_crisis_dir, mock_detector):
mgr = CrisisRoomManager()
mgr.check_message("!room1:matrix.org", "@user:matrix.org", "I want to kill myself")
events = mgr.get_room_events("!room1:matrix.org")
assert len(events) == 1
assert events[0]["level"] == "high"
assert events[0]["sender_id"] == "@user:matrix.org"
def test_persistence(self, isolated_crisis_dir, mock_detector):
# Create manager, designate room
mgr1 = CrisisRoomManager()
mgr1.designate_crisis_room("!persist:matrix.org")
# Reset singleton and reload
import agent.crisis_room
agent.crisis_room._manager = None
mgr2 = CrisisRoomManager()
rooms = mgr2.list_crisis_rooms()
assert len(rooms) == 1
assert rooms[0]["room_id"] == "!persist:matrix.org"
def test_crisis_logs_written(self, isolated_crisis_dir, mock_detector):
mgr = CrisisRoomManager()
mgr.check_message("!room1:matrix.org", "@user:matrix.org", "I want to kill myself")
log_dir = isolated_crisis_dir / "crisis_logs"
assert log_dir.exists()
log_files = list(log_dir.glob("crisis_*.jsonl"))
assert len(log_files) >= 1
# Check log content
lines = log_files[0].read_text().strip().split("\n")
assert len(lines) == 1
entry = json.loads(lines[0])
assert entry["level"] == "high"

View File

@@ -0,0 +1,194 @@
"""Tests for time-aware cron model routing — Issue #317."""
import pytest
from datetime import datetime
from agent.smart_model_routing import resolve_cron_model, _hour_in_window
class TestHourInWindow:
"""Hour-in-window detection including midnight wrap."""
def test_normal_window(self):
assert _hour_in_window(18, 17, 22) is True
assert _hour_in_window(16, 17, 22) is False
assert _hour_in_window(22, 17, 22) is False # [start, end) exclusive end
def test_midnight_wrap(self):
assert _hour_in_window(23, 22, 6) is True
assert _hour_in_window(3, 22, 6) is True
assert _hour_in_window(10, 22, 6) is False
def test_edge_cases(self):
assert _hour_in_window(0, 0, 24) is True
assert _hour_in_window(23, 0, 24) is True
assert _hour_in_window(0, 22, 6) is True
assert _hour_in_window(5, 22, 6) is True
assert _hour_in_window(6, 22, 6) is False
class TestResolveCronModel:
"""Time-aware model resolution for cron jobs."""
def _config(self, **overrides):
base = {
"enabled": True,
"fallback_model": "anthropic/claude-sonnet-4",
"fallback_provider": "openrouter",
"windows": [
{"start_hour": 17, "end_hour": 22, "reason": "evening_error_peak"},
],
}
base.update(overrides)
return base
def test_disabled_returns_base_model(self):
result = resolve_cron_model(
"xiaomi/mimo-v2-pro",
{"enabled": False},
now=datetime(2026, 4, 12, 18, 0),
)
assert result["model"] == "xiaomi/mimo-v2-pro"
assert result["overridden"] is False
def test_no_config_returns_base_model(self):
result = resolve_cron_model("xiaomi/mimo-v2-pro", None)
assert result["model"] == "xiaomi/mimo-v2-pro"
assert result["overridden"] is False
def test_no_windows_returns_base_model(self):
result = resolve_cron_model(
"xiaomi/mimo-v2-pro",
{"enabled": True, "windows": []},
now=datetime(2026, 4, 12, 18, 0),
)
assert result["overridden"] is False
def test_evening_window_overrides(self):
"""18:00 falls in [17, 22) — should override to stronger model."""
result = resolve_cron_model(
"xiaomi/mimo-v2-pro",
self._config(),
now=datetime(2026, 4, 12, 18, 0),
)
assert result["model"] == "anthropic/claude-sonnet-4"
assert result["provider"] == "openrouter"
assert result["overridden"] is True
assert "evening_error_peak" in result["reason"]
assert "hour=18" in result["reason"]
def test_outside_window_keeps_base(self):
"""09:00 is outside [17, 22) — keep base model."""
result = resolve_cron_model(
"xiaomi/mimo-v2-pro",
self._config(),
now=datetime(2026, 4, 12, 9, 0),
)
assert result["model"] == "xiaomi/mimo-v2-pro"
assert result["overridden"] is False
def test_window_boundary_start_inclusive(self):
"""17:00 is start of window — should override."""
result = resolve_cron_model(
"xiaomi/mimo-v2-pro",
self._config(),
now=datetime(2026, 4, 12, 17, 0),
)
assert result["overridden"] is True
def test_window_boundary_end_exclusive(self):
"""22:00 is end of window — should NOT override."""
result = resolve_cron_model(
"xiaomi/mimo-v2-pro",
self._config(),
now=datetime(2026, 4, 12, 22, 0),
)
assert result["overridden"] is False
def test_midnight_window(self):
"""Overnight window [22, 6) wraps midnight."""
config = self._config(windows=[
{"start_hour": 22, "end_hour": 6, "reason": "overnight_instability"},
])
# 23:00 — in window
result = resolve_cron_model("mimo", config, now=datetime(2026, 4, 12, 23, 0))
assert result["overridden"] is True
assert "overnight_instability" in result["reason"]
# 03:00 — in window (past midnight)
result = resolve_cron_model("mimo", config, now=datetime(2026, 4, 13, 3, 0))
assert result["overridden"] is True
# 10:00 — outside window
result = resolve_cron_model("mimo", config, now=datetime(2026, 4, 12, 10, 0))
assert result["overridden"] is False
def test_per_window_model_override(self):
"""Window-specific model takes precedence over global fallback."""
config = self._config(windows=[
{
"start_hour": 17,
"end_hour": 22,
"model": "anthropic/claude-opus-4-6",
"provider": "anthropic",
"reason": "peak_hours",
},
])
result = resolve_cron_model("mimo", config, now=datetime(2026, 4, 12, 18, 0))
assert result["model"] == "anthropic/claude-opus-4-6"
assert result["provider"] == "anthropic"
def test_first_matching_window_wins(self):
"""When windows overlap, first match wins."""
config = self._config(windows=[
{"start_hour": 17, "end_hour": 20, "model": "strong-1", "provider": "p1", "reason": "w1"},
{"start_hour": 19, "end_hour": 22, "model": "strong-2", "provider": "p2", "reason": "w2"},
])
# 19:00 matches both — first wins
result = resolve_cron_model("mimo", config, now=datetime(2026, 4, 12, 19, 0))
assert result["model"] == "strong-1"
def test_no_fallback_model_configured(self):
"""If no fallback_model, keeps base model even in window."""
config = {"enabled": True, "windows": [
{"start_hour": 17, "end_hour": 22, "reason": "test"},
]}
result = resolve_cron_model("mimo", config, now=datetime(2026, 4, 12, 18, 0))
assert result["overridden"] is False
assert result["model"] == "mimo"
def test_malformed_windows_skipped(self):
"""Non-dict or missing hours in windows are skipped safely."""
config = self._config(windows=[
"not-a-dict",
{"start_hour": 17}, # missing end_hour
{"end_hour": 22}, # missing start_hour
{"start_hour": "bad", "end_hour": "bad"},
{"start_hour": 17, "end_hour": 22, "reason": "valid"},
])
result = resolve_cron_model("mimo", config, now=datetime(2026, 4, 12, 18, 0))
assert result["overridden"] is True
assert "valid" in result["reason"]
def test_empty_provider_defaults_to_empty(self):
"""When window has no provider, falls back to global, then empty."""
config = self._config(
fallback_provider="",
windows=[{"start_hour": 17, "end_hour": 22, "reason": "test"}],
)
result = resolve_cron_model("mimo", config, now=datetime(2026, 4, 12, 18, 0))
assert result["provider"] == ""
def test_multiple_windows_coverage(self):
"""Two non-overlapping windows cover evening and overnight."""
config = self._config(windows=[
{"start_hour": 17, "end_hour": 22, "reason": "evening"},
{"start_hour": 2, "end_hour": 5, "reason": "overnight"},
])
# Evening
assert resolve_cron_model("mimo", config, now=datetime(2026, 4, 12, 20, 0))["overridden"] is True
# Overnight
assert resolve_cron_model("mimo", config, now=datetime(2026, 4, 13, 3, 0))["overridden"] is True
# Safe hours
assert resolve_cron_model("mimo", config, now=datetime(2026, 4, 12, 10, 0))["overridden"] is False
assert resolve_cron_model("mimo", config, now=datetime(2026, 4, 12, 1, 0))["overridden"] is False

View File

@@ -1,131 +0,0 @@
"""Crisis Room Tool — manage Matrix crisis rooms with SOUL.md protocol.
Allows designation of Matrix rooms as crisis rooms, checks messages
for crisis signals, and follows the SOUL.md 'When a Man Is Dying'
protocol.
"""
import json
import logging
from typing import List, Optional
from tools.registry import registry
logger = logging.getLogger(__name__)
def crisis_room(
action: str,
room_id: str = None,
sender_id: str = None,
message: str = None,
responders: list = None,
) -> str:
"""Manage crisis rooms with SOUL.md protocol.
Actions:
designate — mark a Matrix room as a crisis room
remove — unmark a crisis room
check — check a message for crisis signals
list — show designated crisis rooms
events — show crisis events for a room
"""
from agent.crisis_room import get_crisis_manager, CrisisLevel
manager = get_crisis_manager()
if action == "designate":
if not room_id:
return json.dumps({"success": False, "error": "room_id is required."})
result = manager.designate_crisis_room(room_id, responders or [])
return json.dumps(result)
if action == "remove":
if not room_id:
return json.dumps({"success": False, "error": "room_id is required."})
result = manager.remove_crisis_room(room_id)
return json.dumps(result)
if action == "check":
if not room_id or not message:
return json.dumps({
"success": False,
"error": "room_id and message are required for 'check'.",
})
result = manager.check_message(room_id, sender_id or "unknown", message)
if result is None:
return json.dumps({"success": True, "crisis_detected": False})
return json.dumps({"success": True, **result})
if action == "list":
rooms = manager.list_crisis_rooms()
return json.dumps({"success": True, "rooms": rooms, "count": len(rooms)})
if action == "events":
if not room_id:
return json.dumps({"success": False, "error": "room_id is required."})
events = manager.get_room_events(room_id)
return json.dumps({"success": True, "events": events, "count": len(events)})
return json.dumps({
"success": False,
"error": f"Unknown action '{action}'. Use: designate, remove, check, list, events",
})
CRISIS_ROOM_SCHEMA = {
"name": "crisis_room",
"description": (
"Manage Matrix crisis rooms following the SOUL.md 'When a Man Is Dying' protocol. "
"Designated crisis rooms are monitored for crisis signals. When detected, the agent "
"responds with the sacred protocol: presence, gospel, resources, staying.\n\n"
"Actions:\n"
" designate — mark a Matrix room as a crisis room (with optional responder list)\n"
" remove — unmark a crisis room\n"
" check — check a message for crisis signals (returns protocol response)\n"
" list — show designated crisis rooms\n"
" events — show crisis event history for a room"
),
"parameters": {
"type": "object",
"properties": {
"action": {
"type": "string",
"enum": ["designate", "remove", "check", "list", "events"],
"description": "The action to perform.",
},
"room_id": {
"type": "string",
"description": "Matrix room ID (e.g. !abc123:matrix.org).",
},
"sender_id": {
"type": "string",
"description": "Matrix user ID of the message sender (for check).",
},
"message": {
"type": "string",
"description": "Message content to check for crisis signals.",
},
"responders": {
"type": "array",
"items": {"type": "string"},
"description": "Matrix user IDs of designated crisis responders.",
},
},
"required": ["action"],
},
}
registry.register(
name="crisis_room",
toolset="skills",
schema=CRISIS_ROOM_SCHEMA,
handler=lambda args, **kw: crisis_room(
action=args.get("action", ""),
room_id=args.get("room_id"),
sender_id=args.get("sender_id"),
message=args.get("message"),
responders=args.get("responders"),
),
emoji="🆘",
)