diff --git a/app/dialog/flows/veteran_onboarding.py b/app/dialog/flows/veteran_onboarding.py new file mode 100644 index 0000000..38104ee --- /dev/null +++ b/app/dialog/flows/veteran_onboarding.py @@ -0,0 +1,458 @@ +"""Veteran start screening (issue #36, CLINICAL_MODULES §1, ADR-001/002/007/010). + +Entered by the core flow after consent + age gate when the profile has module `veteran`: + + veteran.onboarding.start intro «что нормально после возвращения» + FAQ ──«Начнём»──▶ + veteran.onboarding.exposure trauma exposure (yes / no / decline) + yes ──▶ scale.pcptsd5 ──▶ ┐ + no / decline ───────────▶ ┤ (PC-PTSD-5 is not administered; nothing is stored for it) + veteran.onboarding.pcl5 ─┘──▶ scale.pcl5 + veteran.onboarding.phq9 ──▶ scale.phq9 (item 9 buttons only; positive → engine crisis path) + veteran.onboarding.auditc sex from profile, asked if unknown ──▶ scale.auditc + veteran.onboarding.weapon weapon access (4 buttons) → facts; access + risk signal → safe-storage block + veteran.onboarding.result plain-language summary, modules enabled in profile → finish_onboarding() + +Clinical decisions encoded here: + * PCL-5 is given to every veteran regardless of the exposure answer: the module's start set is PCL-5/PHQ-9/ + AUDIT-C (CLINICAL_MODULES §1); PC-PTSD-5 is only a short entry screen and, by its own instructions, is asked + only after a "yes" to trauma exposure (`config/scales/pcptsd5.yaml: notes`). + * Module `depression` is enabled when the PHQ-9 total meets `config/modules.yaml: + veteran.enables_depression_if` (≥ 10). Scores and severity labels are never shown to the patient. + * Safe-storage block (means restriction, one of the best-evidenced suicide-prevention measures) is shown when + the patient reports access to a weapon AND there is a risk signal in this onboarding: a scale flag listed in + `settings.weapon_risk_flags` (PHQ-9 item 9 ≥ 1, C-SSRS, AUDIT-C positive), a safety event (POSSIBLE/ACUTE) + or a crisis opened since the onboarding began. Deterministic, no LLM (ADR-007). + +Facts for lane B (`dialog_state.data["facts"]`, merged with whatever else is there): + trauma_exposure True | False | None (declined) + weapon_access True (self / others_or_safe) | False (none) | None (declined) + weapon_access_choice "none" | "self" | "others_or_safe" | "declined" + weapon_access_at ISO timestamp of the answer + weapon_safety_block True when the safe-storage block was shown +`PatientSnapshot.weapon_access` should be built from `facts["weapon_access"]`. + +Crisis during onboarding: the engine moves to `crisis` and drops the running scale session. Progress is kept in +`data["veteran_onboarding"]` (`stage`, `since_id`); when the FSM later enters `idle` (doctor acknowledged the +crisis) the onboarding resumes at the stored stage, and scale stages whose result was already stored since the +onboarding began are skipped (`since_id` = last scale_results id before the start). +""" + +from __future__ import annotations + +from functools import lru_cache +from pathlib import Path +from typing import Any + +import yaml +from pydantic import BaseModel, ConfigDict, Field + +from app.config import get_settings +from app.core.modules import load_modules +from app.db import repo +from app.db.clinical import ScaleResult +from app.dialog.contracts import Button, Event, Reply +from app.dialog.crisis import crisis_open +from app.dialog.flows.core import finish_onboarding, menu_reply +from app.dialog.fsm import Ctx, Step, on_enter, state, trigger +from app.dialog.scale_session import start_scale + +MODULE = "veteran" +PREFIX = "veteran.onboarding" +START = f"{PREFIX}.start" +EXPOSURE = f"{PREFIX}.exposure" +PCL5 = f"{PREFIX}.pcl5" +PHQ9 = f"{PREFIX}.phq9" +AUDITC = f"{PREFIX}.auditc" +WEAPON = f"{PREFIX}.weapon" +RESULT = f"{PREFIX}.result" + +BTN = "vonb" # button data namespace: "vonb::" +FAQ_KEYS = ("who", "unit", "licences") + +# choice code → value of facts["weapon_access"]: storage with others still means the patient owns a weapon. +WEAPON_ACCESS: dict[str, bool | None] = {"none": False, "self": True, "others_or_safe": True, "declined": None} +EXPOSURE_VALUE: dict[str, bool | None] = {"yes": True, "no": False, "declined": None} +SEX_VALUE: dict[str, str | None] = {"m": "m", "f": "f", "declined": None} + + +# ------------------------------------------------------------------------------------------- texts + + +class _Strict(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True) + + +class OnbSettings(_Strict): + weapon_risk_flags: list[str] = Field(default_factory=list) + + +class IntroTexts(_Strict): + text: str + go: str + repeat: str + resume: str + + +class ExposureTexts(_Strict): + question: str + answer_yes: str + answer_no: str + decline: str + ack_yes: str + ack_other: str + + +class SexTexts(_Strict): + question: str + male: str + female: str + decline: str + + +class WeaponTexts(_Strict): + question: str + no_access: str + self: str + others_or_safe: str + decline: str + ack: str + safety_block_self: str + safety_block_stored: str + safety_block_unknown: str + + +class ResultTexts(_Strict): + intro: str + calm: str + load: str + ptsd: str + mood: str + alcohol: str + next: str + + +class FaqEntry(_Strict): + button: str + text: str + + +class FaqTexts(_Strict): + who: FaqEntry + unit: FaqEntry + licences: FaqEntry + + +class VeteranOnboardingTexts(_Strict): + status: str + settings: OnbSettings + intro: IntroTexts + buttons_hint: str + exposure: ExposureTexts + sex: SexTexts + weapon: WeaponTexts + result: ResultTexts + faq: FaqTexts + + +def texts_path() -> Path: + return get_settings().config_dir / "texts" / "veteran_onboarding.yaml" + + +@lru_cache +def load_texts(path: Path | None = None) -> VeteranOnboardingTexts: + path = path or texts_path() + return VeteranOnboardingTexts.model_validate(yaml.safe_load(path.read_text(encoding="utf-8"))) + + +# ------------------------------------------------------------------------------------------- helpers + + +def _progress(ctx: Ctx) -> dict[str, Any]: + return dict(ctx.data.get("veteran_onboarding") or {}) + + +def _at_stage(ctx: Ctx, stage: str) -> dict[str, Any]: + """Data update remembering the current stage (resume point after a crisis).""" + return {"veteran_onboarding": {**_progress(ctx), "stage": stage}} + + +def _facts(ctx: Ctx, **updates: Any) -> dict[str, Any]: + return {"facts": {**(ctx.data.get("facts") or {}), **updates}} + + +def _faq_buttons() -> list[list[Button]]: + faq = load_texts().faq + return [[Button(label=getattr(faq, key).button, data=f"{BTN}:faq:{key}")] for key in FAQ_KEYS] + + +def _choice(event: Event, question: str, allowed: dict[str, Any]) -> str | None: + if event.type != "button" or not event.button: + return None + parts = event.button.split(":") + if len(parts) == 3 and parts[0] == BTN and parts[1] == question and parts[2] in allowed: + return parts[2] + return None + + +def _faq_key(event: Event) -> str | None: + if event.type == "button" and event.button and event.button.startswith(f"{BTN}:faq:"): + key = event.button.rsplit(":", 1)[1] + return key if key in FAQ_KEYS else None + return None + + +async def _results_since(ctx: Ctx) -> list[ScaleResult]: + since_id = int(_progress(ctx).get("since_id") or 0) + return [r for r in await repo.list_scale_results(ctx.session, ctx.token) if r.id > since_id] + + +async def _latest(ctx: Ctx, scale: str) -> ScaleResult | None: + rows = [r for r in await _results_since(ctx) if r.scale == scale] + return rows[-1] if rows else None + + +async def _risk_signals(ctx: Ctx) -> list[str]: + """Deterministic risk signals collected during this onboarding (see module docstring).""" + risk_flags = set(load_texts().settings.weapon_risk_flags) + signals = sorted({f for r in await _results_since(ctx) for f in (r.flags or {}) if f in risk_flags}) + started = str(_progress(ctx).get("started_at") or "") + events = [ + e + for e in ctx.data.get("safety_events") or [] + if str(e.get("ts", "")) >= started and e.get("level") in ("possible", "acute") + ] + if events: + signals.append("safety_event") + opened_at = str((ctx.data.get("crisis") or {}).get("opened_at") or "") + if crisis_open(ctx.data) or (opened_at and opened_at >= started): + signals.append("crisis") + return signals + + +async def _scale_stage(ctx: Ctx, stage: str, scale: str, next_stage: str) -> Step: + """Start `scale` unless its result is already stored in this onboarding (resume after a crisis).""" + if await _latest(ctx, scale) is not None: + return Step(next_state=next_stage, data=_at_stage(ctx, next_stage)) + step = await start_scale(ctx, scale, return_state=next_stage, scheduled=True) + step.data.update(_at_stage(ctx, stage)) + return step + + +# ------------------------------------------------------------------------------------------- start + + +def _intro_reply(text: str) -> Reply: + go = Button(label=load_texts().intro.go, data=f"{BTN}:go:yes") + return Reply(text=text, buttons=[[go], *_faq_buttons()], kind="question") + + +@on_enter(START) +async def _start_enter(ctx: Ctx) -> Step: + rows = await repo.list_scale_results(ctx.session, ctx.token) + progress = { + "stage": START, + "since_id": max((r.id for r in rows), default=0), + "started_at": ctx.now.isoformat(), + } + return Step(replies=[_intro_reply(load_texts().intro.text)], data={"veteran_onboarding": progress}) + + +# ------------------------------------------------------------------------------------------- questions + + +def _exposure_reply(note: str = "") -> Reply: + t = load_texts().exposure + buttons = [ + [Button(label=t.answer_yes, data=f"{BTN}:exposure:yes")], + [Button(label=t.answer_no, data=f"{BTN}:exposure:no")], + [Button(label=t.decline, data=f"{BTN}:exposure:declined")], + ] + return Reply(text="\n\n".join(p for p in (note, t.question) if p), buttons=buttons, kind="question") + + +def _sex_reply(note: str = "") -> Reply: + t = load_texts().sex + buttons = [ + [Button(label=t.male, data=f"{BTN}:sex:m")], + [Button(label=t.female, data=f"{BTN}:sex:f")], + [Button(label=t.decline, data=f"{BTN}:sex:declined")], + ] + return Reply(text="\n\n".join(p for p in (note, t.question) if p), buttons=buttons, kind="question") + + +def _weapon_reply(note: str = "") -> Reply: + t = load_texts().weapon + buttons = [ + [Button(label=t.no_access, data=f"{BTN}:weapon:none")], + [Button(label=t.self, data=f"{BTN}:weapon:self")], + [Button(label=t.others_or_safe, data=f"{BTN}:weapon:others_or_safe")], + [Button(label=t.decline, data=f"{BTN}:weapon:declined")], + ] + return Reply(text="\n\n".join(p for p in (note, t.question) if p), buttons=buttons, kind="question") + + +def _current_question(ctx: Ctx, note: str = "") -> Reply: + if ctx.state == EXPOSURE: + return _exposure_reply(note) + if ctx.state == AUDITC: + return _sex_reply(note) + if ctx.state == WEAPON: + return _weapon_reply(note) + t = load_texts().intro + return _intro_reply("\n\n".join(p for p in (note, t.repeat) if p)) + + +@on_enter(EXPOSURE) +async def _exposure_enter(ctx: Ctx) -> Step: + return Step(replies=[_exposure_reply()], data=_at_stage(ctx, EXPOSURE)) + + +@on_enter(PCL5) +async def _pcl5_enter(ctx: Ctx) -> Step: + return await _scale_stage(ctx, PCL5, "pcl5", PHQ9) + + +@on_enter(PHQ9) +async def _phq9_enter(ctx: Ctx) -> Step: + # Item 9 is buttons-only and a positive answer opens the crisis path inside the scale session (engine). + return await _scale_stage(ctx, PHQ9, "phq9", AUDITC) + + +@on_enter(AUDITC) +async def _auditc_enter(ctx: Ctx) -> Step: + if await _latest(ctx, "auditc") is not None: + return Step(next_state=WEAPON, data=_at_stage(ctx, WEAPON)) + profile = await ctx.profile() + known = profile is not None and profile.sex in ("m", "f") + if known or _progress(ctx).get("sex_asked"): + # Unknown sex after "не хочу отвечать": AUDIT-C uses the more sensitive default threshold. + return await _scale_stage(ctx, AUDITC, "auditc", WEAPON) + return Step(replies=[_sex_reply()], data=_at_stage(ctx, AUDITC)) + + +@on_enter(WEAPON) +async def _weapon_enter(ctx: Ctx) -> Step: + return Step(replies=[_weapon_reply()], data=_at_stage(ctx, WEAPON)) + + +@state(PREFIX) +async def _handle(ctx: Ctx, event: Event) -> Step: + """Events in the onboarding's own question states (scales are handled by the scale session).""" + t = load_texts() + faq = _faq_key(event) + if faq is not None: + return Step(replies=[Reply(text=getattr(t.faq, faq).text), _current_question(ctx)]) + if event.type == "system": + return Step() # reminders do not interrupt the screening + + if ctx.state == START: + if _choice(event, "go", {"yes": True}): + return Step(next_state=EXPOSURE) + return Step(replies=[_current_question(ctx)]) + + if ctx.state == EXPOSURE and (code := _choice(event, "exposure", EXPOSURE_VALUE)): + exposed = EXPOSURE_VALUE[code] + step = Step(data=_facts(ctx, trauma_exposure=exposed)) + if exposed: + step.replies.append(Reply(text=t.exposure.ack_yes)) + nxt = await start_scale(ctx, "pcptsd5", return_state=PCL5, scheduled=True) + nxt.data.update(_at_stage(ctx, EXPOSURE)) + return step.then(nxt) + step.replies.append(Reply(text=t.exposure.ack_other)) + return step.then(Step(next_state=PCL5)) + + if ctx.state == AUDITC and (code := _choice(event, "sex", SEX_VALUE)): + sex = SEX_VALUE[code] + if sex is not None: + await repo.upsert_profile(ctx.session, token=ctx.token, sex=sex) + step = await _scale_stage(ctx, AUDITC, "auditc", WEAPON) + step.data["veteran_onboarding"] = {**step.data["veteran_onboarding"], "sex_asked": True} + return step + + if ctx.state == WEAPON and (code := _choice(event, "weapon", WEAPON_ACCESS)): + return await _weapon_answer(ctx, code) + + return Step(replies=[_current_question(ctx, t.buttons_hint)]) + + +async def _weapon_answer(ctx: Ctx, code: str) -> Step: + t = load_texts().weapon + access = WEAPON_ACCESS[code] + replies = [Reply(text=t.ack)] + show_block = access is not False and bool(await _risk_signals(ctx)) + if show_block: + block = {"self": t.safety_block_self, "others_or_safe": t.safety_block_stored}.get(code, t.safety_block_unknown) + replies = [Reply(text=block, kind="safety_plan")] + data = _facts( + ctx, + weapon_access=access, + weapon_access_choice=code, + weapon_access_at=ctx.now.isoformat(), + weapon_safety_block=show_block, + ) + # notify_monitoring: weapon access is a monitoring fact (lane B rule "доступ к оружию → блок безопасности"). + return Step(replies=replies, next_state=RESULT, data=data, notify_monitoring=True) + + +# ------------------------------------------------------------------------------------------- result + + +def _meets(total: int, op: str, value: int) -> bool: + return total >= value if op == ">=" else total > value + + +@on_enter(RESULT) +async def _result_enter(ctx: Ctx) -> Step: + t = load_texts().result + pcl5 = await _latest(ctx, "pcl5") + auditc = await _latest(ctx, "auditc") + + # Depression module: PHQ-9 total vs. the threshold from config/modules.yaml (veteran.enables_depression_if). + rule = load_modules().modules[MODULE].enables_depression_if + trigger_result = await _latest(ctx, rule.scale) if rule is not None else None + depression = bool(rule and trigger_result and _meets(trigger_result.total, rule.op, rule.value)) + profile = await ctx.profile() + modules = list(profile.modules or []) if profile is not None else [] + for module in (MODULE, "depression") if depression else (MODULE,): + if module not in modules: + modules.append(module) + await repo.upsert_profile(ctx.session, token=ctx.token, modules=modules) + + lines = [] + if pcl5 is not None and "pcl5_probable_ptsd" in (pcl5.flags or {}): + lines.append(t.ptsd) + if depression: + lines.append(t.mood) + if auditc is not None and "auditc_positive" in (auditc.flags or {}): + lines.append(t.alcohol) + body = [t.intro, t.load, *lines] if lines else [t.intro, t.calm] + summary = Reply(text="\n".join([*body, "", t.next]), buttons=_faq_buttons()) + done = Step( + replies=[summary], + data={"veteran_onboarding": {**_progress(ctx), "stage": None, "done_at": ctx.now.isoformat()}}, + notify_monitoring=True, + ) + return done.then(await finish_onboarding(ctx, module=MODULE)) + + +# ------------------------------------------------------------------------------------------- resume / FAQ + + +@on_enter("idle") +async def _resume_after_crisis(ctx: Ctx) -> Step: + """Back in idle with an unfinished veteran onboarding (it was interrupted by a crisis) → continue it.""" + stage = _progress(ctx).get("stage") + if not stage or not str(stage).startswith(PREFIX) or crisis_open(ctx.data) or ctx.data.get("consent") is not True: + return Step() + stage = EXPOSURE if stage == START else str(stage) + return Step(replies=[Reply(text=load_texts().intro.resume)], next_state=stage) + + +def _faq_trigger(key: str) -> None: + @trigger(f"button:{BTN}:faq:{key}") + async def _faq(ctx: Ctx, event: Event) -> Step: + return Step(replies=[menu_reply(ctx, getattr(load_texts().faq, key).text)]) + + +for _key in FAQ_KEYS: + _faq_trigger(_key) diff --git a/config/texts/veteran_onboarding.yaml b/config/texts/veteran_onboarding.yaml new file mode 100644 index 0000000..09eeee3 --- /dev/null +++ b/config/texts/veteran_onboarding.yaml @@ -0,0 +1,92 @@ +# Тексты стартового онбординга ветерана (issue #36): вступление, вопрос о травматическом опыте, пол для AUDIT-C, +# вопрос о доступе к оружию, блок безопасного хранения, итог, FAQ. +# СТАТУС: черновик, утверждает психиатр (#8). Ответы FAQ — формулировку проверяет юрист. +# +# Правила (проверяются тестом tests/test_dialog_veteran_onboarding.py): +# * тон — как у сослуживца: на «вы», коротко, без жалости и без нравоучений; +# * без слов «диагноз», «расстройство», «психолог», «психиатр», «болезнь», «псих»; +# * о сервисе — никогда «учёт», «контроль», «надзор» (страх пациента можно процитировать только в подписи +# кнопки FAQ — так человек узнаёт свой вопрос); +# * пациент не видит баллов и названий шкал. + +status: "черновик, утверждает психиатр (#8)" + +settings: + # Флаги шкал (config/scales/*.yaml), которые считаются сигналом риска для блока безопасного хранения оружия. + # Положительный п.9 PHQ-9 и C-SSRS — суицидальный риск; AUDIT-C — алкоголь как фактор риска при доступе к оружию. + weapon_risk_flags: [phq9_item9_positive, cssrs_high, cssrs_moderate, auditc_positive] + +intro: + text: |- + Рад, что вы здесь. Пара слов, прежде чем начнём. + + После возвращения многие замечают одно и то же: сон сбит, в голову лезет прошлое, тело всё ещё «на дежурстве», раздражает то, что раньше не трогало, тянет выпить, чтобы отпустило. Это частая реакция на то, через что вы прошли, а не слабость — и с этим реально работают. + + Сейчас будет несколько коротких вопросов про сон, напряжение, настроение и алкоголь — минут пять, почти всё кнопками. Ответы увидит только ваш лечащий врач: так он сможет помогать точнее. + go: "Начнём" + repeat: "Когда будете готовы — нажмите «Начнём». Если есть вопросы, ответы на частые — кнопками ниже." + resume: "Давайте продолжим с того места, где остановились. Осталось немного." + +buttons_hint: "Пожалуйста, выберите вариант кнопкой ниже." + +exposure: + question: |- + Бывало ли с вами событие, которое сильно напугало, потрясло или было по-настоящему опасным для жизни — на службе, в бою или в другое время? + Подробности рассказывать не нужно. + answer_yes: "Да" + answer_no: "Нет" + decline: "Не хочу отвечать" + ack_yes: "Понял. Несколько вопросов о том, как это даёт о себе знать сейчас." + ack_other: "Хорошо, идём дальше." + +sex: + question: "Один уточняющий вопрос: ваш пол? Он нужен только для того, чтобы правильно учесть ответы про алкоголь — нормы для мужчин и женщин разные." + male: "Мужской" + female: "Женский" + decline: "Не хочу отвечать" + +weapon: + question: |- + Последний вопрос, он про безопасность, а не про проверку: есть ли у вас сейчас доступ к оружию — табельному, охотничьему, наградному или любому другому? + no_access: "Нет" + self: "Есть, хранится у меня" + others_or_safe: "Есть, у других / в сейфе" + decline: "Не хочу отвечать" + ack: "Спасибо, что ответили." + # Блок безопасного хранения — фиксированный текст, показывается только при доступе к оружию И сигнале риска. + safety_block_self: |- + Одна просьба — без нравоучений. Пока вам тяжело, пусть оружие и патроны побудут у того, кому вы доверяете: у брата, друга, в сейфе, ключ от которого у другого человека. Дело не в недоверии к вам. В трудную минуту важно, чтобы между мыслью и действием было время — это одна из самых надёжных мер защиты, и так делают многие. + Это можно спокойно обсудить с врачом на приёме. + safety_block_stored: |- + Хорошо, что оружие хранится не под рукой. Пусть пока так и остаётся: ключ или код — у человека, которому вы доверяете, а патроны — отдельно. В трудную минуту важно, чтобы между мыслью и действием было время. + Это можно спокойно обсудить с врачом на приёме. + safety_block_unknown: |- + Понимаю, что не хочется об этом говорить. Просто на всякий случай: если дома есть оружие или большой запас лекарств, пусть пока они побудут у человека, которому вы доверяете. В трудную минуту важно, чтобы между мыслью и действием было время. + +result: + intro: "Спасибо, что ответили на всё. Коротко, что видно по ответам:" + calm: "Сейчас серьёзных трудностей по ответам не видно. Хорошо — будем держать руку на пульсе." + load: "Нагрузка сейчас заметная. Так бывает у многих после возвращения, и это то, с чем врач может помочь." + ptsd: "• Прошлое пока не отпускает: напряжение, воспоминания, сон." + mood: "• Настроение и силы ниже, чем хотелось бы." + alcohol: "• Алкоголь — стоит обсудить с врачом, особенно если назначат препараты: они с алкоголем плохо сочетаются." + next: "Ваши ответы уже у лечащего врача — он обсудит их с вами на приёме. А я буду на связи каждый день с одним коротким вопросом и напоминанием о препарате." + +faq: + who: + button: "Кто увидит мои ответы?" + # формулировку проверяет юрист + text: |- + Только ваш лечащий врач — тот, кто дал вам код подключения. Перед автоматической обработкой из сообщений удаляются имена, телефоны и другие личные данные. Данные хранятся в России. + Отозвать согласие и остановить обработку можно в любой момент — кнопкой «Отозвать согласие» или командой /revoke. + unit: + button: "Это попадёт на учёт / в часть?" + # формулировку проверяет юрист + text: |- + Нет. Это добровольная консультативная помощь между визитами к врачу. Ваши ответы не передаются в часть, военкомат, работодателю или кому-то ещё без вашего согласия — кроме случаев, прямо предусмотренных законом. Отказ от помощника никак не влияет на ваше лечение. + licences: + button: "Заберут права на оружие или вождение?" + # формулировку проверяет юрист + text: |- + Помощник не выдаёт заключений и не отправляет ваши ответы ни в ГИБДД, ни в Росгвардию, ни на медкомиссии. Допуск к оружию и вождению решает отдельная медицинская комиссия по своим правилам, установленным законом. + Как именно это связано с вашим лечением — спросите лечащего врача прямо. Это нормальный вопрос, его задают многие, и честный ответ лучше, чем откладывать помощь из-за неизвестности. diff --git a/tests/test_dialog_veteran_onboarding.py b/tests/test_dialog_veteran_onboarding.py new file mode 100644 index 0000000..a181158 --- /dev/null +++ b/tests/test_dialog_veteran_onboarding.py @@ -0,0 +1,368 @@ +"""Veteran start screening (#36): demo persona Andrey end-to-end through `Engine.handle` on the offline mock LLM. + +Connect code → consent → age (known) → intro → trauma exposure → PC-PTSD-5 → PCL-5 → PHQ-9 → AUDIT-C → weapon +question → plain-language result → idle. In-memory SQLite (identity / clinical), fake monitoring (lane B stub). +""" + +import re +from collections.abc import AsyncIterator +from pathlib import Path +from typing import Any + +import pytest +import yaml + +from app.core import clock +from app.core.rules import Action +from app.db import identity_repo, repo, session +from app.db.bootstrap import init_db +from app.dialog.contracts import Event, Reply +from app.dialog.engine import Engine +from app.dialog.flows import veteran_onboarding as vo +from app.dialog.texts import load_dialog_texts +from app.integrations.base import VerifiedPerson, integrations +from app.llm.gateway import LLMGateway +from app.llm.mock import MockTasks + +TOKEN = "b" * 32 +MEM = "sqlite+aiosqlite:///:memory:" +TEXTS = load_dialog_texts() +VT = vo.load_texts() + +# Andrey's screening (docs/product/01_demo_persona.md): PC-PTSD-5 = 4, PCL-5 = 46, PHQ-9 = 14 (item 9 = 0), +# AUDIT-C = 3 (below the male threshold), no weapon at home. +PCPTSD5 = [1, 1, 1, 1, 0] +PCL5 = [3] * 6 + [2] * 14 +PHQ9 = [2, 2, 2, 2, 2, 2, 1, 1, 0] +AUDITC = [2, 1, 0] + + +class FakeMonitoring: + def __init__(self) -> None: + self.calls = 0 + + async def after_data_change(self, token: str) -> list[Action]: + self.calls += 1 + return [] + + async def is_in_crisis(self, token: str) -> bool: + return True # only doctor_ack ends a crisis in these tests + + +async def _make_patient(sex: str | None) -> None: + async with session.get_identity_session() as s: + await identity_repo.create_patient( + s, + token=TOKEN, + display_name="Андрей (синтетика)", + snils="000-000-000 00", + clinic_id="c1", + doctor_id="doc-test", + now=clock.now(), + birth_year=1988, + sex=sex or "m", + ) + async with session.get_clinical_session() as s: + await repo.upsert_profile(s, token=TOKEN, doctor_id="doc-test", modules=["veteran"], sex=sex) + + +@pytest.fixture +async def dbs() -> AsyncIterator[None]: + await session.configure(identity_url=MEM, clinical_url=MEM) + await init_db() + await _make_patient("m") + yield + await session.configure() + + +@pytest.fixture +async def dbs_nosex() -> AsyncIterator[None]: + await session.configure(identity_url=MEM, clinical_url=MEM) + await init_db() + await _make_patient(None) + yield + await session.configure() + + +@pytest.fixture +def mon() -> FakeMonitoring: + return FakeMonitoring() + + +@pytest.fixture +def engine(mon: FakeMonitoring) -> Engine: + return Engine(gateway=LLMGateway(MockTasks()), monitoring=mon) + + +# ----------------------------------------------------------------------------------------------- helpers + + +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)) + + +def buttons(reply: Reply) -> list[str]: + return [b.data for row in reply.buttons for b in row] + + +def scale_buttons(replies: list[Reply], scale: str) -> list[str]: + for reply in reversed(replies): + found = [d for d in buttons(reply) if d.startswith(f"scale:{scale}:")] + if found: + return found + raise AssertionError(f"no question of {scale} in replies: {[r.text[:40] for r in replies]}") + + +async def answer(e: Engine, replies: list[Reply], scale: str, values: list[int]) -> list[Reply]: + """Answer `values` item by item with the buttons of the currently shown question.""" + for value in values: + choice = [d for d in scale_buttons(replies, scale) if d.endswith(f":{value}")] + assert choice, f"{scale}: no option {value}" + replies = await press(e, choice[0]) + return replies + + +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 results() -> dict[str, Any]: + async with session.get_clinical_session() as s: + return {r.scale: r for r in await repo.list_scale_results(s, TOKEN)} + + +async def profile() -> Any: + async with session.get_clinical_session() as s: + return await repo.get_profile(s, TOKEN) + + +async def to_intro(e: Engine) -> list[Reply]: + await e.handle(TOKEN, Event(type="start")) + return await press(e, "consent:yes") + + +async def to_weapon(e: Engine, phq9: list[int] = PHQ9) -> list[Reply]: + await to_intro(e) + r = await press(e, "vonb:go:yes") + assert buttons(r[-1]) == ["vonb:exposure:yes", "vonb:exposure:no", "vonb:exposure:declined"] + r = await press(e, "vonb:exposure:yes") + r = await answer(e, r, "pcptsd5", PCPTSD5) + r = await answer(e, r, "pcl5", PCL5) + r = await answer(e, r, "phq9", phq9) + return r + + +# ----------------------------------------------------------------------------------------------- happy path + + +async def test_andrey_full_onboarding( + dbs: None, engine: Engine, mon: FakeMonitoring, monkeypatch: pytest.MonkeyPatch +) -> None: + class Ident: + async def verify(self, auth_code: str) -> VerifiedPerson | None: + return VerifiedPerson(token=TOKEN, doctor_id="doc-test") if auth_code == "QR-ANDREY" else None + + monkeypatch.setattr(integrations, "identity", Ident()) + r = await engine.handle(TOKEN, Event(type="start", params={"code": "QR-ANDREY"})) + assert buttons(r[0]) == ["consent:yes", "consent:no"] + + r = await press(engine, "consent:yes") # birth year known → adult → veteran onboarding + assert (await get_state())[0] == vo.START + assert r[-1].text == VT.intro.text + assert buttons(r[-1]) == ["vonb:go:yes", "vonb:faq:who", "vonb:faq:unit", "vonb:faq:licences"] + + r = await press(engine, "vonb:go:yes") + assert (await get_state())[0] == vo.EXPOSURE + r = await press(engine, "vonb:exposure:yes") + assert (await get_state())[0] == "scale.pcptsd5" + r = await answer(engine, r, "pcptsd5", PCPTSD5) + assert (await get_state())[0] == "scale.pcl5" + r = await answer(engine, r, "pcl5", PCL5) + assert (await get_state())[0] == "scale.phq9" + r = await answer(engine, r, "phq9", PHQ9) + assert (await get_state())[0] == "scale.auditc" # sex known from profile → no question + r = await answer(engine, r, "auditc", AUDITC) + assert (await get_state())[0] == vo.WEAPON + weapon_choices = ["none", "self", "others_or_safe", "declined"] + assert buttons(r[-1]) == [f"vonb:weapon:{c}" for c in weapon_choices] + + r = await press(engine, "vonb:weapon:none") + state, data = await get_state() + assert state == "idle" + texts = [x.text for x in r] + assert VT.weapon.ack in texts + summary = next(x for x in r if x.text.startswith(VT.result.intro)) + assert VT.result.ptsd in summary.text and VT.result.mood in summary.text + assert VT.result.alcohol not in summary.text # AUDIT-C 3 < 4 for men + assert not re.search(r"\d", summary.text) # no scores shown to the patient + assert r[-1].text == TEXTS.idle.welcome and "menu:checkin" in buttons(r[-1]) + + res = await results() + assert {k: v.total for k, v in res.items()} == {"pcptsd5": 4, "pcl5": 46, "phq9": 14, "auditc": 3} + assert res["phq9"].item9 == 0 and res["auditc"].severity == "negative" + assert all(v.complete and v.scheduled for v in res.values()) + assert (await profile()).modules == ["veteran", "depression"] + + facts = data["facts"] + assert facts["trauma_exposure"] is True + assert facts["weapon_access"] is False and facts["weapon_access_choice"] == "none" + assert facts["weapon_safety_block"] is False and "weapon_access_at" in facts + assert data["veteran_onboarding"].get("stage") is None and "done_at" in data["veteran_onboarding"] + assert "crisis" not in data + assert mon.calls >= 5 # every scale + weapon + result + + +async def test_exposure_no_skips_pcptsd5(dbs: None, engine: Engine) -> None: + await to_intro(engine) + await press(engine, "vonb:go:yes") + r = await press(engine, "vonb:exposure:no") + assert (await get_state())[0] == "scale.pcl5" # veterans get PCL-5 regardless + assert VT.exposure.ack_other in [x.text for x in r] + r = await answer(engine, r, "pcl5", [0] * 20) + r = await answer(engine, r, "phq9", [0] * 9) + r = await answer(engine, r, "auditc", [0, 0, 0]) + r = await press(engine, "vonb:weapon:declined") + state, data = await get_state() + assert state == "idle" + assert "pcptsd5" not in await results() + assert data["facts"]["trauma_exposure"] is False + assert data["facts"]["weapon_access"] is None and data["facts"]["weapon_access_choice"] == "declined" + assert any(VT.result.calm in x.text for x in r) + assert (await profile()).modules == ["veteran"] # PHQ-9 0 < 10 → no depression module + + +async def test_exposure_declined_and_free_text_reasks(dbs: None, engine: Engine) -> None: + await to_intro(engine) + await press(engine, "vonb:go:yes") + r = await text(engine, "ну такое") + assert (await get_state())[0] == vo.EXPOSURE + assert VT.buttons_hint in r[-1].text and "vonb:exposure:yes" in buttons(r[-1]) + await press(engine, "vonb:exposure:declined") + state, data = await get_state() + assert state == "scale.pcl5" and data["facts"]["trauma_exposure"] is None + + +async def test_auditc_asks_sex_when_unknown(dbs_nosex: None, engine: Engine) -> None: + await to_weapon(engine) + assert (await get_state())[0] == vo.AUDITC + r = await press(engine, "vonb:sex:f") + assert (await get_state())[0] == "scale.auditc" + assert (await profile()).sex == "f" + await answer(engine, r, "auditc", AUDITC) + res = await results() + assert res["auditc"].severity == "positive" # 3 is positive for women + + +async def test_auditc_sex_declined_uses_default_threshold(dbs_nosex: None, engine: Engine) -> None: + await to_weapon(engine) + r = await press(engine, "vonb:sex:declined") + assert (await profile()).sex is None + r = await answer(engine, r, "auditc", AUDITC) + assert (await results())["auditc"].severity == "positive" # unknown sex → sensitive threshold ≥ 3 + # Weapon at home + AUDIT-C positive = risk signal → safe-storage block. + r = await press(engine, "vonb:weapon:self") + assert VT.weapon.safety_block_self in [x.text for x in r] + + +# ----------------------------------------------------------------------------------------------- weapon / safety + + +async def test_weapon_without_risk_no_block(dbs: None, engine: Engine) -> None: + r = await to_weapon(engine) + r = await answer(engine, r, "auditc", AUDITC) + r = await press(engine, "vonb:weapon:others_or_safe") + texts = [x.text for x in r] + assert VT.weapon.safety_block_stored not in texts and VT.weapon.safety_block_self not in texts + _, data = await get_state() + assert data["facts"]["weapon_access"] is True and data["facts"]["weapon_access_choice"] == "others_or_safe" + assert data["facts"]["weapon_safety_block"] is False + + +async def test_item9_positive_crisis_then_resume_and_safety_block(dbs: None, engine: Engine) -> None: + r = await to_weapon(engine, phq9=[2, 2, 2, 2, 2, 2, 1, 1, 1]) + state, data = await get_state() + # Engine crisis path: fixed script, C-SSRS queued and started while in crisis. + assert data["crisis"]["open"] is True and data["crisis"]["source"] == "scale:phq9" + assert any(x.kind == "crisis" for x in r) + assert state == "scale.cssrs_screen" + r = await answer(engine, r, "cssrs_screen", [0] * 6) + assert (await get_state())[0] == "crisis" + + # Doctor acknowledged → idle → the onboarding resumes where it stopped (PHQ-9 already stored → AUDIT-C). + r = await engine.handle(TOKEN, Event(type="system", system="doctor_ack")) + assert VT.intro.resume in [x.text for x in r] + assert (await get_state())[0] == "scale.auditc" + r = await answer(engine, r, "auditc", AUDITC) + r = await press(engine, "vonb:weapon:self") + state, data = await get_state() + assert VT.weapon.safety_block_self in [x.text for x in r] + assert data["facts"]["weapon_access"] is True and data["facts"]["weapon_safety_block"] is True + assert state == "idle" + assert (await results())["phq9"].item9 == 1 + + +# ----------------------------------------------------------------------------------------------- FAQ + + +async def test_faq_in_onboarding_and_idle(dbs: None, engine: Engine) -> None: + await to_intro(engine) + for key in vo.FAQ_KEYS: + r = await press(engine, f"vonb:faq:{key}") + assert r[0].text == getattr(VT.faq, key).text + assert "vonb:go:yes" in buttons(r[-1]) # the intro is shown again + assert (await get_state())[0] == vo.START + await press(engine, "vonb:go:yes") + r = await press(engine, "vonb:faq:unit") + assert r[0].text == VT.faq.unit.text and "vonb:exposure:yes" in buttons(r[-1]) + + # FAQ buttons keep working after onboarding (idle triggers). + r = await answer(engine, await press(engine, "vonb:exposure:no"), "pcl5", [0] * 20) + r = await answer(engine, r, "phq9", [0] * 9) + r = await answer(engine, r, "auditc", [0, 0, 0]) + await press(engine, "vonb:weapon:none") + assert (await get_state())[0] == "idle" + r = await press(engine, "vonb:faq:licences") + assert r[0].text == VT.faq.licences.text and "menu:checkin" in buttons(r[0]) + + +def test_faq_texts_honest() -> None: + assert "прямо предусмотренных законом" in VT.faq.unit.text + assert "лечащий врач" in VT.faq.who.text + raw = Path(vo.texts_path()).read_text(encoding="utf-8") + assert raw.count("формулировку проверяет юрист") >= 3 + assert VT.status == "черновик, утверждает психиатр (#8)" + + +# ----------------------------------------------------------------------------------------------- texts scan + +FORBIDDEN = ["диагноз", "расстройств", "психолог", "психиатр", "болезн", "псих"] +FORBIDDEN_SERVICE = ["учёт", "учет", "контрол", "надзор"] + + +def _strings(node: Any, path: str = "") -> list[tuple[str, str]]: + if isinstance(node, dict): + return [s for k, v in node.items() for s in _strings(v, f"{path}.{k}")] + if isinstance(node, list): + return [s for v in node for s in _strings(v, path)] + return [(path, node)] if isinstance(node, str) else [] + + +def test_yaml_forbidden_words() -> None: + data = yaml.safe_load(Path(vo.texts_path()).read_text(encoding="utf-8")) + strings = [(p, s) for p, s in _strings(data) if p != ".status"] + assert strings + for path, value in strings: + low = value.lower() + for word in FORBIDDEN: + assert word not in low, f"{path}: «{word}»" + # Quoting the patient's own fear in a FAQ button label is allowed; statements about the service are not. + if not (path.startswith(".faq.") and path.endswith(".button")): + for word in FORBIDDEN_SERVICE: + assert word not in low, f"{path}: «{word}»"