ElebobaandClaude Opus 5.5
d06720fdab
CI / ruff · pytest (pull_request) Successful in 1m19s
- No same-day C-SSRS repeat: a risk screen completed within cssrs_cooldown_hours is not queued again after PHQ-9 item 9 >= 1 (scale crisis path, rule schedule_scale, scale_due); crisis script and alerts stay. - Scale intro registry: register_scale_intro(scale_id, text_or_fn); start_scale(intro=None) uses it. veteran_daily migrated from per-scale on_enter hooks. - Prefix triggers "button:<prefix>:*" with trigger_payload(event); medication buttons carry the dose (med:taken:<dose_id>, ...), med_pending dropped; payload-less legacy buttons answer as stale. - Abandoned-step timeout: settings.abandon_after_hours (default 6 h) returns abandonable states (daily.*, meds.*, scale.* not in onboarding) to idle before the next event; never in crisis or with intake on. Refs #56 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
378 lines
17 KiB
Python
378 lines
17 KiB
Python
"""Medication flow and check-in units (#37): reminder buttons, skip reasons, side effects, medication question,
|
||
med_info rule reactions, check-in edge cases, text policy of the YAML files."""
|
||
|
||
from collections.abc import AsyncIterator
|
||
from datetime import datetime, timedelta
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
import pytest
|
||
import yaml
|
||
from sqlalchemy import select
|
||
|
||
from app.config import Settings, get_settings
|
||
from app.core import clock
|
||
from app.core.rules import Action
|
||
from app.db import repo, session
|
||
from app.db.bootstrap import init_db
|
||
from app.db.clinical import Dose, Prescription
|
||
from app.dialog.contracts import Event, Reply
|
||
from app.dialog.engine import Engine
|
||
from app.dialog.flows.meds import load_meds_texts
|
||
from app.dialog.flows.veteran_daily import load_daily_texts
|
||
from app.dialog.fsm import Ctx, enter_hook_for
|
||
from app.dialog.scale_session import start_scale
|
||
from app.dialog.texts import load_dialog_texts
|
||
from app.integrations.base import DrugSection, integrations
|
||
from app.llm import med_qa as med_qa_module
|
||
from app.llm.gateway import LLMGateway
|
||
from app.llm.med_qa import fallback_text
|
||
from app.llm.mock import MockTasks
|
||
|
||
TOKEN = "m" * 32
|
||
MEM = "sqlite+aiosqlite:///:memory:"
|
||
NOW = datetime(2026, 10, 7, 8, 0)
|
||
MEDS = load_meds_texts()
|
||
DAILY = load_daily_texts()
|
||
SECTIONS = {
|
||
"side_effects": "Тошнота обычно возникает в первые дни приёма и проходит в течение 1–2 недель.",
|
||
"onset": "Улучшение обычно отмечается через 2–4 недели регулярного приёма.",
|
||
"alcohol": "Совместный приём с алкоголем не рекомендуется.",
|
||
"administration": "Препарат принимают один раз в сутки, утром или вечером, независимо от еды.",
|
||
}
|
||
|
||
|
||
class FakeDrugInfo:
|
||
async def sections(self, inn: str) -> list[DrugSection]:
|
||
return [DrugSection(inn=inn, section=k, text=v, source="синтетика") for k, v in SECTIONS.items()]
|
||
|
||
|
||
class ScriptedMonitoring:
|
||
def __init__(self) -> None:
|
||
self.calls = 0
|
||
self.actions: list[Action] = []
|
||
|
||
async def after_data_change(self, token: str) -> list[Action]:
|
||
self.calls += 1
|
||
out, self.actions = self.actions, []
|
||
return out
|
||
|
||
async def is_in_crisis(self, token: str) -> bool:
|
||
return False
|
||
|
||
|
||
@pytest.fixture
|
||
async def mon(monkeypatch: pytest.MonkeyPatch) -> AsyncIterator[ScriptedMonitoring]:
|
||
monkeypatch.setattr(clock, "get_settings", lambda: Settings(_env_file=None, demo_clock=True))
|
||
monkeypatch.setattr(integrations, "drug_info", FakeDrugInfo())
|
||
monkeypatch.setattr(med_qa_module, "get_gateway", lambda: LLMGateway(MockTasks()))
|
||
clock.set_now(NOW)
|
||
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="doc", modules=["veteran"], sex="m", birth_year=1990, therapy_start=NOW.date()
|
||
)
|
||
s.add(Prescription(token=TOKEN, drug="сертралин", dose="50 мг", schedule={"times": ["08:00"]}))
|
||
await repo.set_dialog_state(s, token=TOKEN, state="idle", data={"consent": True}, now=clock.now())
|
||
yield ScriptedMonitoring()
|
||
clock.reset()
|
||
await session.configure()
|
||
|
||
|
||
@pytest.fixture
|
||
def engine(mon: ScriptedMonitoring) -> Engine:
|
||
return Engine(gateway=LLMGateway(MockTasks()), monitoring=mon)
|
||
|
||
|
||
async def press(e: Engine, data: str) -> list[Reply]:
|
||
return await e.handle(TOKEN, Event(type="button", button=data))
|
||
|
||
|
||
async def text(e: Engine, t: str) -> list[Reply]:
|
||
return await e.handle(TOKEN, Event(type="text", text=t))
|
||
|
||
|
||
async def system(e: Engine, kind: str, **params: Any) -> list[Reply]:
|
||
return await e.handle(TOKEN, Event(type="system", system=kind, params=params)) # type: ignore[arg-type]
|
||
|
||
|
||
def buttons(replies: list[Reply]) -> list[str]:
|
||
return [b.data for r in replies for row in r.buttons for b in row]
|
||
|
||
|
||
def kinds(replies: list[Reply]) -> list[str]:
|
||
"""Button data without the dose payload (#56): "med:taken:7" → "med:taken"."""
|
||
return [d.rsplit(":", 1)[0] if d.startswith("med:") else d for d in buttons(replies)]
|
||
|
||
|
||
async def tap(e: Engine, replies: list[Reply], kind: str) -> list[Reply]:
|
||
"""Press the button of `replies` whose data (without payload) is `kind` — as the patient would."""
|
||
[data] = [d for d in buttons(replies) if d.startswith("med:") and d.rsplit(":", 1)[0] == kind]
|
||
return await press(e, data)
|
||
|
||
|
||
async def get_state() -> tuple[str, dict[str, Any]]:
|
||
async with session.get_clinical_session() as s:
|
||
row = await repo.get_dialog_state(s, TOKEN)
|
||
assert row is not None
|
||
return row.state, dict(row.data)
|
||
|
||
|
||
async def checkin() -> Any:
|
||
async with session.get_clinical_session() as s:
|
||
rows = await repo.list_checkins(s, TOKEN)
|
||
return rows[-1] if rows else None
|
||
|
||
|
||
async def doses() -> list[Dose]:
|
||
async with session.get_clinical_session() as s:
|
||
return list((await s.scalars(select(Dose).where(Dose.token == TOKEN).order_by(Dose.id))).all())
|
||
|
||
|
||
# ----------------------------------------------------------------------------------------------- reminder
|
||
|
||
|
||
async def test_reminder_without_dose_id_creates_dose_and_taken(engine: Engine, mon: ScriptedMonitoring) -> None:
|
||
r = await system(engine, "med_reminder")
|
||
assert r[0].text == MEDS.reminder.text.format(drug="сертралин", dose="50 мг")
|
||
[dose] = await doses()
|
||
assert buttons(r) == [f"med:taken:{dose.id}", f"med:skip:{dose.id}", f"med:side_effects:{dose.id}"]
|
||
assert dose.status == "pending"
|
||
r = await tap(engine, r, "med:taken")
|
||
assert r[0].text == MEDS.taken_ack and mon.calls == 1
|
||
[dose] = await doses()
|
||
assert dose.status == "taken" and dose.answered_at is not None
|
||
assert (await checkin()).med_taken is True
|
||
|
||
|
||
async def test_stale_reminder_button(engine: Engine) -> None:
|
||
reminder = await system(engine, "med_reminder")
|
||
clock.advance(days=1)
|
||
r = await tap(engine, reminder, "med:taken")
|
||
assert r[0].text == MEDS.stale and await checkin() is None
|
||
|
||
|
||
@pytest.mark.parametrize("reason", ["forgot", "other"])
|
||
async def test_skip_simple_reasons(engine: Engine, reason: str) -> None:
|
||
r = await tap(engine, await system(engine, "med_reminder"), "med:skip")
|
||
assert kinds(r) == [f"med:skip_reason:{x}" for x in ("forgot", "side_effects", "quit", "other")]
|
||
r = await tap(engine, r, f"med:skip_reason:{reason}")
|
||
assert r[0].text == MEDS.skip.ack[reason]
|
||
c = await checkin()
|
||
assert c.med_taken is False and not c.quit_intent
|
||
assert [d.status for d in await doses()] == ["skipped"]
|
||
|
||
|
||
async def test_skip_quit_sets_intent_and_explains(engine: Engine, mon: ScriptedMonitoring) -> None:
|
||
skip = await tap(engine, await system(engine, "med_reminder"), "med:skip")
|
||
mon.actions = [Action(type="med_info", code="quit_intent", params={"topic": "delayed_effect"})]
|
||
r = await tap(engine, skip, "med:skip_reason:quit")
|
||
assert r[0].text == MEDS.skip.ack["quit"]
|
||
assert "2–4 недели" in r[1].text and MEDS.med_info["delayed_effect"].notice in r[1].text
|
||
assert (await checkin()).quit_intent is True
|
||
|
||
|
||
async def test_skip_because_of_side_effects_asks_which(engine: Engine) -> None:
|
||
skip = await tap(engine, await system(engine, "med_reminder"), "med:skip")
|
||
r = await tap(engine, skip, "med:skip_reason:side_effects")
|
||
assert "med:se:nausea" in kinds(r) and "med:se:other" in kinds(r)
|
||
|
||
|
||
async def test_side_effect_button(engine: Engine, mon: ScriptedMonitoring) -> None:
|
||
r = await tap(engine, await system(engine, "med_reminder"), "med:side_effects")
|
||
assert kinds(r) == [f"med:se:{c}" for c in ("nausea", "insomnia", "headache", "dizziness", "agitation", "other")]
|
||
r = await tap(engine, r, "med:se:nausea")
|
||
assert "проходит в течение 1–2 недель" in r[0].text # instruction excerpt
|
||
assert r[1].text == MEDS.side_effects.ask_dose and kinds(r[1:]) == ["med:dose:yes", "med:dose:no"]
|
||
assert (await checkin()).side_effects == ["nausea"]
|
||
assert (await doses())[0].side_effects == "nausea"
|
||
r = await tap(engine, r, "med:dose:yes")
|
||
assert (await checkin()).med_taken is True and (await doses())[0].status == "taken"
|
||
|
||
|
||
async def test_side_effect_other_free_text(engine: Engine) -> None:
|
||
reminder = await system(engine, "med_reminder")
|
||
r = await tap(engine, await tap(engine, reminder, "med:side_effects"), "med:se:other")
|
||
assert r[0].text == MEDS.side_effects.describe and (await get_state())[0] == "meds.side_effect_text"
|
||
r = await text(engine, "голова болит весь день")
|
||
assert (await checkin()).side_effects == ["headache"] and (await get_state())[0] == "idle"
|
||
await tap(engine, await tap(engine, reminder, "med:side_effects"), "med:se:other")
|
||
await text(engine, "какое-то странное ощущение")
|
||
assert (await checkin()).side_effects == ["headache", "other"]
|
||
|
||
|
||
async def test_same_side_effect_as_yesterday_gets_support(engine: Engine) -> None:
|
||
async with session.get_clinical_session() as s:
|
||
await repo.upsert_checkin(
|
||
s, token=TOKEN, day=clock.today() - timedelta(days=1), now=clock.now(), side_effects=["nausea"]
|
||
)
|
||
se = await tap(engine, await system(engine, "med_reminder"), "med:side_effects")
|
||
r = await tap(engine, se, "med:se:nausea")
|
||
assert r[0].text == MEDS.side_effects.repeat
|
||
|
||
|
||
# ----------------------------------------------------------------------------------------------- question menu
|
||
|
||
|
||
async def test_med_question_menu(engine: Engine) -> None:
|
||
r = await press(engine, "menu:med_question")
|
||
assert r[0].text == MEDS.question.prompt and (await get_state())[0] == "meds.question"
|
||
r = await text(engine, "можно ли пить алкоголь?")
|
||
assert "Совместный приём с алкоголем не рекомендуется" in r[0].text and (await get_state())[0] == "idle"
|
||
await press(engine, "menu:med_question")
|
||
r = await text(engine, "можно увеличить дозу до 100 мг?")
|
||
assert r[0].text == fallback_text()
|
||
|
||
|
||
async def test_med_question_menu_left_by_button(engine: Engine) -> None:
|
||
await press(engine, "menu:med_question")
|
||
r = await press(engine, "menu:checkin")
|
||
assert r[0].text == DAILY.checkin.start and (await get_state())[0] == "daily.mood"
|
||
|
||
|
||
async def test_med_question_without_prescription(engine: Engine) -> None:
|
||
async with session.get_clinical_session() as s:
|
||
for row in await s.scalars(select(Prescription)):
|
||
await s.delete(row)
|
||
r = await press(engine, "menu:med_question")
|
||
assert r[0].text == MEDS.question.no_prescription and (await get_state())[0] == "idle"
|
||
|
||
|
||
async def test_med_info_without_excerpt_uses_fixed_text(
|
||
engine: Engine, mon: ScriptedMonitoring, monkeypatch: pytest.MonkeyPatch
|
||
) -> None:
|
||
class Empty:
|
||
async def sections(self, inn: str) -> list[DrugSection]:
|
||
return []
|
||
|
||
monkeypatch.setattr(integrations, "drug_info", Empty())
|
||
mon.actions = [Action(type="med_info", code="alcohol_3d", params={"topic": "alcohol_interaction"})]
|
||
r = await press(engine, "menu:visit") # any data-writing step runs monitoring
|
||
spec = MEDS.med_info["alcohol_interaction"]
|
||
assert r[0].text == DAILY.visit.requested
|
||
assert r[1].text == "\n\n".join([spec.lead, spec.fallback, spec.notice])
|
||
assert (await get_state())[1]["visit_requests"]
|
||
|
||
|
||
async def test_idle_free_text_med_question_and_side_effect(engine: Engine, mon: ScriptedMonitoring) -> None:
|
||
r = await text(engine, "когда подействуют таблетки?")
|
||
assert "2–4 недели" in r[0].text and "menu:checkin" in buttons(r)
|
||
r = await text(engine, "подташнивает после таблеток")
|
||
assert "проходит в течение 1–2 недель" in r[0].text and (await checkin()).side_effects == ["nausea"]
|
||
assert mon.calls == 1
|
||
|
||
|
||
# ----------------------------------------------------------------------------------------------- check-in edges
|
||
|
||
|
||
async def test_checkin_asks_dose_when_not_answered_and_typed_mood(engine: Engine) -> None:
|
||
await system(engine, "daily_checkin")
|
||
r = await text(engine, "отлично")
|
||
assert DAILY.checkin.buttons_hint in r[0].text and (await get_state())[0] == "daily.mood"
|
||
r = await text(engine, "7")
|
||
assert r[0].text == MEDS.checkin.question.format(drug="сертралин")
|
||
await press(engine, "daily:med:no")
|
||
c = await checkin()
|
||
assert (c.mood, c.med_taken) == (7, False)
|
||
assert (await get_state())[0] == "daily.sleep"
|
||
|
||
|
||
async def test_reminder_during_checkin_keeps_checkin(engine: Engine) -> None:
|
||
await system(engine, "daily_checkin")
|
||
r = await system(engine, "med_reminder")
|
||
assert kinds(r[:1]) == ["med:taken", "med:skip", "med:side_effects"] and r[-1].text == DAILY.checkin.mood.question
|
||
r = await tap(engine, r, "med:taken")
|
||
assert r[0].text == MEDS.taken_ack and (await get_state())[0] == "daily.mood"
|
||
await press(engine, "daily:mood:5")
|
||
assert (await get_state())[0] == "daily.sleep" # dose already answered → no med question
|
||
|
||
|
||
async def test_unfinished_checkin_dropped_next_day(engine: Engine) -> None:
|
||
await system(engine, "daily_checkin")
|
||
await press(engine, "daily:mood:2")
|
||
clock.advance(days=1)
|
||
r = await system(engine, "daily_checkin")
|
||
assert r[0].text == DAILY.checkin.start and (await get_state())[0] == "daily.mood"
|
||
|
||
|
||
async def test_checkin_once_a_day(engine: Engine) -> None:
|
||
await system(engine, "daily_checkin")
|
||
for b in ("daily:mood:6", "daily:med:yes", "daily:sleep:6-8", "daily:nightmares:no", "daily:alcohol:none"):
|
||
await press(engine, b)
|
||
await press(engine, "daily:text:skip")
|
||
assert await system(engine, "daily_checkin") == []
|
||
r = await press(engine, "menu:checkin")
|
||
assert r[0].text == DAILY.checkin.already_done
|
||
|
||
|
||
async def test_alcohol_clarify_for_non_veteran(engine: Engine) -> None:
|
||
async with session.get_clinical_session() as s:
|
||
await repo.upsert_profile(s, token=TOKEN, modules=["depression"])
|
||
await system(engine, "daily_checkin")
|
||
await press(engine, "daily:mood:5")
|
||
r = await press(engine, "daily:med:yes")
|
||
assert buttons(r) == ["daily:text:skip"] # no sleep / alcohol questions outside the veteran module
|
||
r = await text(engine, "вечером выпил пива с друзьями")
|
||
assert r[-1].text == DAILY.checkin.alcohol.clarify and (await get_state())[0] == "daily.alcohol_amount"
|
||
r = await press(engine, "daily:alcohol_amount:more")
|
||
assert r[0].text == DAILY.checkin.done and (await checkin()).alcohol_bucket == "more"
|
||
|
||
|
||
async def test_buttons_win_over_llm(engine: Engine) -> None:
|
||
await system(engine, "daily_checkin")
|
||
for b in ("daily:mood:5", "daily:med:yes", "daily:sleep:6-8", "daily:nightmares:no", "daily:alcohol:none"):
|
||
await press(engine, b)
|
||
await text(engine, "почти не спал, кошмары, брошу таблетки")
|
||
c = await checkin()
|
||
assert (c.sleep_bucket, c.nightmares, c.alcohol_bucket, c.quit_intent) == ("6-8", False, "none", True)
|
||
assert c.free_text_scrubbed
|
||
|
||
|
||
async def test_scale_intro_not_added_to_onboarding(engine: Engine, mon: ScriptedMonitoring) -> None:
|
||
await system(engine, "scale_due", scale="dar5", scheduled=True)
|
||
state, data = await get_state()
|
||
assert state == "scale.dar5" and data["scales"]["dar5"]["intro"] == DAILY.scale_intros.scheduled["dar5"]
|
||
|
||
async with session.get_clinical_session() as s:
|
||
ctx = Ctx(
|
||
token=TOKEN,
|
||
state="idle",
|
||
data={},
|
||
now=clock.now(),
|
||
session=s,
|
||
gateway=LLMGateway(MockTasks()),
|
||
monitoring=mon,
|
||
texts=load_dialog_texts(),
|
||
)
|
||
step = await start_scale(ctx, "pcl5", return_state="veteran.onboarding.phq9")
|
||
ctx.data.update(step.data)
|
||
ctx.state = "scale.pcl5"
|
||
hook = enter_hook_for("scale.pcl5")
|
||
assert hook is not None
|
||
entered = await hook(ctx)
|
||
assert DAILY.scale_intros.scheduled["pcl5"] not in entered.replies[0].text
|
||
|
||
|
||
# ----------------------------------------------------------------------------------------------- text policy
|
||
|
||
|
||
def _strings(node: Any) -> list[str]:
|
||
if isinstance(node, str):
|
||
return [node]
|
||
if isinstance(node, dict):
|
||
return [s for v in node.values() for s in _strings(v)]
|
||
if isinstance(node, list):
|
||
return [s for v in node for s in _strings(v)]
|
||
return []
|
||
|
||
|
||
@pytest.mark.parametrize("name", ["meds.yaml", "veteran_daily.yaml"])
|
||
def test_texts_policy(name: str) -> None:
|
||
path: Path = get_settings().config_dir / "texts" / name
|
||
raw = path.read_text(encoding="utf-8")
|
||
assert "черновик, утверждает психиатр (#8)" in raw.splitlines()[2]
|
||
values = " ".join(_strings(yaml.safe_load(raw))).lower()
|
||
for word in ("учёт", "учет", "контрол", "надзор", "диагноз", "птср", "депресси", "увеличьте доз", "мг"):
|
||
assert word not in values, (name, word)
|