PTSRaiAgent/tests/test_services_monitoring.py

"""MonitoringService (#38): snapshot from the clinical DB → real rules → alerts, colour cache, audit, notify."""

from collections.abc import AsyncIterator
from datetime import datetime, timedelta
from typing import Any

import pytest
from sqlalchemy import select

from app.config import Settings
from app.core import clock
from app.core.types import ScrubbedText
from app.db import repo, session
from app.db.audit import AuditLog
from app.db.bootstrap import init_db
from app.db.clinical import Dose, Prescription
from app.integrations.base import MemoryNotifier, integrations
from app.services import contracts, monitoring
from app.services.snapshot import build_snapshot

TOKEN = "a" * 32
DOCTOR = "doc-smirnova"
MEM = "sqlite+aiosqlite:///:memory:"
DAY0 = datetime(2026, 10, 5, 8, 0)

# Persona (docs/product/01_demo_persona.md): day → (dose taken, mood, sleep, nightmares, alcohol, side effects,
# quit intent). Same table as tests/test_dialog_veteran_days.py, written straight to the DB.
TABLE: dict[int, tuple[bool, int, str, bool, str, list[str], bool]] = {
    1: (True, 4, "4-6", True, "none", [], False),
    2: (True, 4, "4-6", True, "none", ["nausea"], False),
    3: (True, 4, "4-6", True, "none", ["nausea"], False),
    4: (True, 3, "4-6", False, "none", ["nausea"], False),
    5: (False, 3, "4-6", False, "none", [], True),
    6: (True, 4, "6-8", False, "none", [], False),
    7: (True, 4, "6-8", False, "none", [], False),
    8: (True, 4, "<4", True, "1-2", [], False),
    9: (True, 4, "<4", True, "more", [], False),
    10: (True, 4, "<4", True, "more", [], False),
    11: (True, 3, "4-6", False, "1-2", [], False),
    12: (True, 3, "4-6", False, "none", [], False),
    13: (True, 2, "<4", True, "none", [], False),
}
COLORS = ["green"] * 3 + ["yellow"] * 2 + ["green"] * 4 + ["yellow"] * 3 + ["red"]


class FailingNotifier:
    async def send(self, *args: Any, **kwargs: Any) -> None:
        raise ConnectionError("channel down")


@pytest.fixture
async def db(monkeypatch: pytest.MonkeyPatch) -> AsyncIterator[MemoryNotifier]:
    monkeypatch.setattr(clock, "get_settings", lambda: Settings(_env_file=None, demo_clock=True))
    notifier = MemoryNotifier()
    monkeypatch.setattr(integrations, "notifier", notifier)
    clock.set_now(DAY0)
    await session.configure(identity_url=MEM, clinical_url=MEM)
    await init_db()
    async with session.get_clinical_session() as s:
        await repo.upsert_profile(
            s,
            token=TOKEN,
            doctor_id=DOCTOR,
            modules=["veteran", "depression"],
            therapy_start=clock.today(),
            birth_year=1992,
            sex="m",
        )
        s.add(Prescription(token=TOKEN, drug="сертралин", dose="50 мг", schedule={"times": ["08:00"]}))
        for scale, total in (("pcl5", 46), ("phq9", 14), ("auditc", 3)):
            await repo.add_scale_result(
                s,
                token=TOKEN,
                scale=scale,
                total=total,
                severity="-",
                complete=True,
                ts=clock.now() - timedelta(days=3),
                item9=0 if scale == "phq9" else None,
            )
        await repo.set_dialog_state(s, token=TOKEN, state="idle", data={}, now=clock.now())
    yield notifier
    clock.register_persistence(None, None)
    clock.reset()
    await session.configure()


async def scale(
    name: str, total: int, *, item9: int | None = None, severity: str = "-", scheduled: bool = True
) -> None:
    async with session.get_clinical_session() as s:
        await repo.add_scale_result(
            s,
            token=TOKEN,
            scale=name,
            total=total,
            severity=severity,
            complete=True,
            ts=clock.now(),
            item9=item9,
            scheduled=scheduled,
        )


async def color() -> str | None:
    async with session.get_clinical_session() as s:
        profile = await repo.get_profile(s, TOKEN)
        assert profile is not None
        return profile.status_color


async def ack_all() -> None:
    async with session.get_clinical_session() as s:
        for a in await repo.list_alerts(s, token=TOKEN, open_only=True):
            await repo.ack_alert(s, alert_id=a.id, doctor_id=DOCTOR, ts=clock.now())
        await monitoring.refresh_status(s, TOKEN)


async def live_day(day: int) -> None:
    """What the dialog writes on one persona day: the dose, an incoming message and the evening check-in."""
    taken, mood, sleep, nightmares, alcohol, effects, quit_intent = TABLE[day]
    clock.set_now(DAY0 + timedelta(days=day))
    async with session.get_clinical_session() as s:
        rx = await s.scalar(select(Prescription))
        assert rx is not None
        s.add(Dose(token=TOKEN, prescription_id=rx.id, planned_at=clock.now(), status="taken" if taken else "skipped"))
        await repo.add_message(s, token=TOKEN, direction="in", text=ScrubbedText("..."), ts=clock.now())
        await repo.upsert_checkin(
            s,
            token=TOKEN,
            day=clock.today(),
            now=clock.now(),
            mood=mood,
            med_taken=taken,
            sleep_bucket=sleep,
            alcohol_bucket=alcohol,
            nightmares=nightmares,
            side_effects=effects,
            quit_intent=quit_intent,
        )


async def test_service_is_registered() -> None:
    assert isinstance(contracts.get_monitoring(), monitoring.RuleMonitoring)


async def test_andrey_13_days(db: MemoryNotifier) -> None:
    mon = monitoring.RuleMonitoring()
    colors: list[str | None] = []
    returned: dict[int, list[str]] = {}
    for day in range(1, 14):
        if day == 6:
            await ack_all()  # the doctor called Andrey after the day-5 alerts
        await live_day(day)
        if day == 7:
            await scale("pcl5", 44)
            await scale("phq9", 13, item9=0)
        actions = await mon.after_data_change(TOKEN)
        if day == 13:
            assert [a.code for a in actions] == ["mood_low_3d"]  # dialog action: unscheduled PHQ-9
            await scale("cssrs_screen", 0, severity="none", scheduled=False)
            await scale("phq9", 18, item9=2, scheduled=False)
            actions += await mon.after_data_change(TOKEN)
        returned[day] = [a.type for a in actions]
        colors.append(await color())

    assert colors == COLORS
    assert "med_info" in returned[5] and "alert" not in returned[5]  # alerts are persisted, not returned
    assert {"med_info", "suggest_visit"} <= set(returned[10])
    assert {"crisis", "show_safety_plan", "schedule_scale"} <= set(returned[13])
    async with session.get_clinical_session() as s:
        alerts = list(await repo.list_alerts(s, token=TOKEN))
        crisis_audit = (await s.scalars(select(AuditLog).where(AuditLog.action == "crisis_acute"))).all()
    codes = [a.code for a in reversed(alerts)]
    assert codes == [
        "side_effect_3d",
        "quit_intent",
        "sleep_lt4_3n",
        "alcohol_3d",
        "phq9_item9",  # same evaluation: YAML order
        "post_crisis_call",  # the crisis leaves a callback task for the doctor on duty
        "phq9_rise",
    ]
    assert len(crisis_audit) == 1 and crisis_audit[0].token == TOKEN and crisis_audit[0].actor == "system"
    acute = [(d, lvl) for d, lvl, _text, _tok in db.sent if lvl == "acute"]
    assert acute == [(DOCTOR, "acute"), (monitoring.ON_DUTY_DOCTOR, "acute")]
    assert all(tok == TOKEN for *_rest, tok in db.sent)
    assert await mon.is_in_crisis(TOKEN)


async def test_running_crisis_is_not_restarted(db: MemoryNotifier) -> None:
    mon = monitoring.RuleMonitoring()
    await scale("phq9", 18, item9=1)
    first = await mon.after_data_change(TOKEN)
    assert any(a.type == "crisis" for a in first)
    clock.advance(hours=1)
    again = await mon.after_data_change(TOKEN)
    assert not any(a.code == "phq9_item9" for a in again)
    async with session.get_clinical_session() as s:
        assert [a.code for a in await repo.list_alerts(s, token=TOKEN, open_only=True) if a.level == "acute"] == [
            "phq9_item9"
        ]
    await ack_all()
    assert not await mon.is_in_crisis(TOKEN)
    assert await color() == "green"


async def test_old_cssrs_high_does_not_restart_crisis(db: MemoryNotifier) -> None:
    mon = monitoring.RuleMonitoring()
    await scale("cssrs_screen", 4, severity="high", scheduled=False)
    assert any(a.type == "crisis" for a in await mon.after_data_change(TOKEN))
    await ack_all()
    clock.advance(days=2)
    assert not any(a.type == "crisis" for a in await mon.after_data_change(TOKEN))


async def test_weapon_access_from_dialog_facts(db: MemoryNotifier) -> None:
    async with session.get_clinical_session() as s:
        await repo.set_dialog_state(
            s, token=TOKEN, state="idle", data={"facts": {"weapon_access": True}}, now=clock.now()
        )
        snap = await build_snapshot(s, TOKEN, clock.now())
    assert snap is not None and snap.weapon_access is True
    await scale("cssrs_screen", 3, severity="moderate", scheduled=False)
    actions = await monitoring.RuleMonitoring().after_data_change(TOKEN)
    assert {a.code for a in actions if a.type == "crisis"} == {"weapon_access_risk"}
    plan = next(a for a in actions if a.type == "show_safety_plan")
    assert plan.params["block"] == "means_restriction"
    assert await color() == "red"


async def test_snapshot_mapping_and_unknown_patient(db: MemoryNotifier) -> None:
    clock.set_now(DAY0 + timedelta(days=4))
    await live_day(4)
    async with session.get_clinical_session() as s:
        snap = await build_snapshot(s, TOKEN, clock.now())
        assert await build_snapshot(s, "f" * 32, clock.now()) is None
    assert snap is not None
    assert (snap.age, snap.sex, snap.therapy_day, snap.modules) == (34, "m", 4, {"veteran", "depression"})
    c = snap.checkins[-1]
    assert (c.sleep_hours_bucket, c.alcohol, c.side_effects) == ("4-6", "none", ["nausea"])
    assert [p.scale for p in snap.scale_results] == ["pcl5", "phq9", "auditc"]
    assert snap.last_patient_message_at is not None and snap.last_patient_message_at.date() == clock.today()
    assert await monitoring.RuleMonitoring().after_data_change("f" * 32) == []


async def test_young_adult_enhanced_monitoring(db: MemoryNotifier) -> None:
    async with session.get_clinical_session() as s:
        await repo.upsert_profile(s, token=TOKEN, birth_year=2004)
    actions = await monitoring.RuleMonitoring().after_data_change(TOKEN)
    assert not any(a.type == "enhanced_monitoring" for a in actions)  # applied by the service itself
    async with session.get_clinical_session() as s:
        profile = await repo.get_profile(s, TOKEN)
        assert profile is not None and profile.enhanced_monitoring is True
        assert not await repo.list_alerts(s, token=TOKEN)
    assert await color() == "green"


async def test_notifier_failure_keeps_alert(db: MemoryNotifier, monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setattr(integrations, "notifier", FailingNotifier())
    await scale("pcl5", 55)
    await monitoring.RuleMonitoring().after_data_change(TOKEN)
    async with session.get_clinical_session() as s:
        assert [(a.code, a.level) for a in await repo.list_alerts(s, token=TOKEN)] == [("pcl5_severe", "moderate")]
    assert await color() == "yellow"


async def test_clock_offset_persists(db: MemoryNotifier) -> None:
    clock.reset()
    await monitoring.init_clock_persistence()
    clock.advance(days=3)
    await monitoring.flush_clock_writes()
    async with session.get_clinical_session() as s:
        assert await repo.load_clock_offset(s) == pytest.approx(timedelta(days=3).total_seconds())

    # "Restart": in-memory offset lost, restored from clock_state.
    clock.register_persistence(None, None)
    clock.reset()
    await monitoring.init_clock_persistence()
    assert clock.offset() == timedelta(days=3)