Compare commits

..
Author SHA1 Message Date
ElebobaandClaude Opus 5.5 4c0836a1be feat(backend): monitoring service — snapshot, rules, alerts, status colour, clock persistence
CI / ruff · pytest (pull_request) Successful in 1m11s
- app/services/snapshot.py: PatientSnapshot from the clinical DB (field mapping, age, therapy day,
  recent C-SSRS risk, weapon access from dialog facts)
- app/services/monitoring.py: RuleMonitoring registered via register_monitoring; persists alert/crisis
  actions, crisis_acute audit, enhanced_monitoring flag, cached status_color, notifies the doctor (and
  the doctor on duty for crises) after commit; a running crisis is not restarted
- demo clock offset persisted in clock_state, restored on scheduler start
- tests: Andrey's 13 days through the service → G G G Y Y G G G G Y Y Y R

Refs #38

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-04 03:13:57 +03:00
Eleboba 20973bef17 Merge pull request 'fix(bot): пробелы движка — повтор C-SSRS, вступления шкал, данные в кнопках, тайм-аут шага' (#57) from fix/56-engine-gaps into dev
CI / ruff · pytest (push) Successful in 1m10s
2026-10-03 19:03:10 +03:00
ElebobaandClaude Opus 5.5 d06720fdab fix(bot): close dialog engine gaps (C-SSRS repeat, scale intros, button payloads, step timeout)
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>
2026-10-03 19:00:47 +03:00
Eleboba 2b7cbb8c2f Merge pull request 'feat(bot): ежедневный чек-ин ветерана, приём лекарств и реакции на правила' (#55) from feat/37-veteran-daily into dev
CI / ruff · pytest (push) Successful in 1m51s
2026-10-03 18:47:52 +03:00
ElebobaandClaude Opus 5.5 2972b330b0 test(bot): Andrey 13-day acceptance scenario and medication flow units
CI / ruff · pytest (pull_request) Successful in 1m3s
Refs #37

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-03 18:45:53 +03:00
ElebobaandClaude Opus 5.5 2b13d54ed7 feat(bot): veteran daily check-in, medication reminders and rule reactions
Daily check-in (mood → dose → sleep/nightmares → alcohol → free text with
extract_checkin, buttons win over the LLM, progressive upsert of
daily_checkins), medication reminder flow (taken / skip with reason /
side effects → doses + daily_checkins), medication question menu via
med_qa, med_info rule reactions from the drug instruction excerpt,
«записаться раньше» request recorded for lane B, friendly intros for
scheduled scales. actions.py: additive med_info handler hook.

Refs #37

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-03 18:45:53 +03:00
Eleboba 4c6e150162 Merge pull request 'feat(questionnaire): revised intake, diaries and follow-up monitoring' (#53) from feat/52-questionnaire into dev
CI / ruff · pytest (push) Successful in 59s
Reviewed-on: #53
2026-10-03 18:41:31 +03:00
Eleboba de5b1e05fb Merge pull request 'feat(bot): онбординг ветерана — скрининг PC-PTSD-5 → PCL-5, PHQ-9, AUDIT-C, вопрос об оружии' (#54) from feat/36-veteran-onboarding into dev
CI / ruff · pytest (push) Successful in 36s
2026-10-03 18:39:39 +03:00
ElebobaandClaude Opus 5.5 0a92266bab feat(bot): veteran onboarding — exposure, PC-PTSD-5, PCL-5, PHQ-9, AUDIT-C, weapon access
CI / ruff · pytest (pull_request) Successful in 32s
Start screening of the veteran module: peer-tone intro, trauma exposure gate for
PC-PTSD-5, PCL-5/PHQ-9/AUDIT-C via scale sessions (sex asked if unknown), weapon
access question stored as facts for lane B with a safe-storage block on risk,
plain-language result, modules enabled in profile, FAQ, resume after a crisis.

Refs #36

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-03 18:37:46 +03:00
21 changed files with 3893 additions and 19 deletions
+16 -4
View File
@@ -6,7 +6,7 @@ are ignored here. While a crisis is open only crisis-relevant actions (crisis, s
the patient: no medication tips or visit offers in the middle of a crisis. the patient: no medication tips or visit offers in the middle of a crisis.
""" """
from collections.abc import Iterable from collections.abc import Awaitable, Callable, Iterable
from typing import Any from typing import Any
from app.core.rules import Action from app.core.rules import Action
@@ -20,6 +20,16 @@ from app.dialog.scale_session import active_scale, add_pending
_MED_INFO_INTENTS = {"delayed_effect": "quit_intent", "alcohol_interaction": "alcohol_info", "nausea": "nausea_info"} _MED_INFO_INTENTS = {"delayed_effect": "quit_intent", "alcohol_interaction": "alcohol_info", "nausea": "nausea_info"}
_CRISIS_SAFE = {"crisis", "show_safety_plan", "schedule_scale"} _CRISIS_SAFE = {"crisis", "show_safety_plan", "schedule_scale"}
# Optional `med_info` answerer registered by a flow (`flows/meds.py`, #37): answers from the drug instruction
# (med_qa). Returning None falls back to the phrase template below.
MedInfoHandler = Callable[[Ctx, Action], Awaitable[Reply | None]]
_med_info_handler: MedInfoHandler | None = None
def register_med_info_handler(handler: MedInfoHandler | None) -> None:
global _med_info_handler
_med_info_handler = handler
def _plan_line(step: Any) -> str: def _plan_line(step: Any) -> str:
if isinstance(step, dict): if isinstance(step, dict):
@@ -57,9 +67,11 @@ async def execute(ctx: Ctx, actions: Iterable[Action]) -> Step:
pending = add_pending(data, scale_id, bool(action.params.get("scheduled", False))) pending = add_pending(data, scale_id, bool(action.params.get("scheduled", False)))
step = step.then(Step(data={"pending_scales": pending})) step = step.then(Step(data={"pending_scales": pending}))
elif action.type == "med_info": elif action.type == "med_info":
intent = _MED_INFO_INTENTS.get(str(action.params.get("topic")), "default") reply = await _med_info_handler(ctx, action) if _med_info_handler is not None else None
text = await ctx.gateway.phrase(intent, {}) if reply is None:
step = step.then(Step(replies=[Reply(text=text, kind="info")])) intent = _MED_INFO_INTENTS.get(str(action.params.get("topic")), "default")
reply = Reply(text=await ctx.gateway.phrase(intent, {}), kind="info")
step = step.then(Step(replies=[reply]))
elif action.type == "suggest_visit": elif action.type == "suggest_visit":
text = await ctx.gateway.phrase("book_visit_offer", {}) text = await ctx.gateway.phrase("book_visit_offer", {})
button = Button(label=ctx.texts.visit.button, data="menu:visit") button = Button(label=ctx.texts.visit.button, data="menu:visit")
+50 -3
View File
@@ -2,6 +2,10 @@
Pipeline for one incoming event (SPEC §3, guardrails #1, #2, #3): Pipeline for one incoming event (SPEC §3, guardrails #1, #2, #3):
0. abandoned step the FSM sits in a question step (check-in, scale) the patient left more than
`settings.abandon_after_hours` ago → back to idle first (stored answers stay; never in
crisis, never in an onboarding chain), then the event is handled as usual.
text event text event
1. scrub() raw text → ScrubbedText. Raw text is never stored, logged or sent anywhere. 1. scrub() raw text → ScrubbedText. Raw text is never stored, logged or sent anywhere.
2. consent gate no valid consent → no LLM, nothing stored; only the local crisis lexicon runs 2. consent gate no valid consent → no LLM, nothing stored; only the local crisis lexicon runs
@@ -25,7 +29,7 @@ Logs carry only the token, states and event codes (guardrail #8).
import asyncio import asyncio
import logging import logging
from dataclasses import dataclass, field from dataclasses import dataclass, field
from datetime import datetime from datetime import datetime, timedelta
from typing import Any from typing import Any
from weakref import WeakValueDictionary from weakref import WeakValueDictionary
@@ -45,7 +49,16 @@ from app.dialog.contracts import Event, Reply, register_engine
from app.dialog.crisis import crisis_open, no_consent_crisis_text, open_crisis from app.dialog.crisis import crisis_open, no_consent_crisis_text, open_crisis
from app.dialog.flows import load_flows from app.dialog.flows import load_flows
from app.dialog.flows.core import close_crisis, revoke from app.dialog.flows.core import close_crisis, revoke
from app.dialog.fsm import NO_CONSENT_STATES, RESTING_STATES, Ctx, Step, apply_data, enter_hook_for, handler_for from app.dialog.fsm import (
NO_CONSENT_STATES,
RESTING_STATES,
Ctx,
Step,
apply_data,
enter_hook_for,
handler_for,
is_abandonable,
)
from app.dialog.texts import DialogTexts, load_dialog_texts from app.dialog.texts import DialogTexts, load_dialog_texts
from app.llm.gateway import LLMGateway, get_gateway from app.llm.gateway import LLMGateway, get_gateway
from app.services import questionnaire as intake_service from app.services import questionnaire as intake_service
@@ -58,6 +71,8 @@ MAX_ENTER_CHAIN = 8
_patient_locks = WeakValueDictionary() _patient_locks = WeakValueDictionary()
# DB unreachable: answer safely instead of failing the channel (connectivity errors only — bugs must surface). # DB unreachable: answer safely instead of failing the channel (connectivity errors only — bugs must surface).
_DB_DOWN = (OSError, OperationalError, InterfaceError) _DB_DOWN = (OSError, OperationalError, InterfaceError)
# dialog_state.data key: when the patient last acted in (or the FSM entered) the current step — abandon timeout.
STEP_AT = "step_at"
@dataclass @dataclass
@@ -109,6 +124,8 @@ class Engine:
state_before = ctx.state state_before = ctx.state
run = _Run() run = _Run()
if self._abandoned(ctx):
await self._apply(ctx, self._abandon(ctx), run)
await self._apply(ctx, await self._route(ctx, event), run) await self._apply(ctx, await self._route(ctx, event), run)
if run.notify: if run.notify:
@@ -126,6 +143,9 @@ class Engine:
await repo.add_message( await repo.add_message(
session, token=token, direction="out", text=scrub(reply.text), ts=ctx.now, fsm_state=ctx.state session, token=token, direction="out", text=scrub(reply.text), ts=ctx.now, fsm_state=ctx.state
) )
if event.type != "system" or ctx.state != state_before:
# System events the step ignores (reminders inside a scale) do not count as the patient being there.
ctx.data[STEP_AT] = ctx.now.isoformat()
await self._save_state(ctx) await self._save_state(ctx)
task_id = event.params.get("questionnaire_task") task_id = event.params.get("questionnaire_task")
if event.type == "system" and task_id and run.replies: if event.type == "system" and task_id and run.replies:
@@ -145,6 +165,24 @@ class Engine:
) )
return run.replies return run.replies
def _abandoned(self, ctx: Ctx) -> bool:
"""The current question step was left unanswered longer than `abandon_after_hours` (app.core.clock)."""
if intake_enabled() or not _consented(ctx) or ctx.state in RESTING_STATES or crisis_open(ctx.data):
return False
hours = ctx.texts.settings.abandon_after_hours
last = ctx.data.get(STEP_AT)
if not hours or not last or ctx.now - datetime.fromisoformat(last) < timedelta(hours=hours):
return False
return is_abandonable(ctx)
def _abandon(self, ctx: Ctx) -> Step:
"""Drop the unfinished step's progress (answers already stored in item_responses / daily_checkins stay;
an unfinished scale gets no scale_results row) and rest in idle so queued scales may start."""
log.info("dialog step abandoned token=%s state=%s", ctx.token, ctx.state)
scale_id = scale_session.active_scale(ctx.state)
scales = {k: v for k, v in (ctx.data.get("scales") or {}).items() if k != scale_id}
return Step(next_state="idle", data={"checkin": None, "scales": scales or None})
async def _route(self, ctx: Ctx, event: Event) -> Step: async def _route(self, ctx: Ctx, event: Event) -> Step:
consented = _consented(ctx) consented = _consented(ctx)
if event.type == "text": if event.type == "text":
@@ -307,10 +345,19 @@ class Engine:
if not pending or ctx.state not in RESTING_STATES: if not pending or ctx.state not in RESTING_STATES:
return Step() return Step()
screen = ctx.texts.settings.risk_screen_scale screen = ctx.texts.settings.risk_screen_scale
dropped: dict[str, Any] = {}
if any(p["scale"] == screen for p in pending) and await scale_session.recent_risk_screen(ctx, screen):
# Item 9 / rule phq9_item9 / scale_due asked for a C-SSRS the patient completed within the cooldown:
# do not repeat the same suicide screen (#56). Crisis script and alerts are unaffected.
log.info("dialog risk screen skipped (cooldown) token=%s", ctx.token)
pending = [p for p in pending if p["scale"] != screen]
dropped = {"pending_scales": pending or None}
if not pending:
return Step(data=dropped)
if ctx.state == "crisis": if ctx.state == "crisis":
pending_screen = [p for p in pending if p["scale"] == screen] pending_screen = [p for p in pending if p["scale"] == screen]
if not pending_screen: if not pending_screen:
return Step() return Step(data=dropped)
item = pending_screen[0] item = pending_screen[0]
else: else:
item = pending[0] item = pending[0]
+465
View File
@@ -0,0 +1,465 @@
"""Medication flow (#37, case scenario «управление приёмом лекарств»). Module-agnostic: every patient with a
prescription gets it, whatever the clinical module.
system:med_reminder {dose_id, drug, dose}
──▶ «Принял» → dose taken, check-in med_taken = true
──▶ «Пропущу» → dose skipped, med_taken = false ──▶ reason:
забыл / побочки (→ side effects) / решил бросить (quit_intent) / другое
──▶ «Есть побочные эффекты» → code buttons (тошнота, бессонница, …) or «Другое» → free text
→ check-in side_effects (+ dose.side_effects) → «А препарат сегодня приняли?»
menu «Вопрос о лекарстве" ──▶ meds.question (free text) ──▶ med_qa (instruction excerpt) or «уточните у врача»
rule action med_info (actions.py hook) ──▶ instruction excerpt for the rule topic + fixed notice
Everything the patient answers is written to the clinical DB (`doses`, `daily_checkins`) and the step asks the
engine to run monitoring (`notify_monitoring`): the rules (lane B, ADR-007) decide alerts and further actions
(quit intent → delayed-onset explanation, missed doses → caring contact). Nothing here decides risk.
Reminder buttons carry the dose they refer to (#56): "med:taken:<ref>", "med:skip:<ref>",
"med:side_effects:<ref>" and the follow-up buttons likewise ("med:se:<code>:<ref>", "med:skip_reason:<r>:<ref>",
"med:dose:yes:<ref>"), matched by prefix triggers. `<ref>` is the `doses.id` of the reminder or, when there is
no dose row (prescription only known to the МИС), the reminder's ISO date. A button of an earlier day's dose is
stale (friendly note, nothing recorded); so are the old payload-less buttons ("med:taken") of earlier releases.
Check-in answers without a button (daily.med step, free text) go to today's planned dose, if any.
"""
from datetime import date, timedelta
from functools import lru_cache
from pathlib import Path
from typing import Any
import yaml
from pydantic import BaseModel, ConfigDict
from sqlalchemy import select
from app.config import get_settings
from app.core.privacy import scrub
from app.core.rules import Action
from app.db import repo
from app.db.clinical import DailyCheckin, Dose
from app.db.clinical import Prescription as PrescriptionRow
from app.dialog.actions import register_med_info_handler
from app.dialog.contracts import Button, Event, Reply
from app.dialog.flows.core import menu_reply
from app.dialog.fsm import Ctx, Step, abandonable, get_trigger, handler_for, state, trigger, trigger_payload
from app.integrations.base import integrations
from app.llm.med_qa import fallback_text, med_qa
QUESTION_STATE = "meds.question"
SIDE_EFFECT_TEXT_STATE = "meds.side_effect_text"
SIDE_EFFECT_CODES = ("nausea", "insomnia", "headache", "dizziness", "agitation", "other")
SKIP_REASONS = ("forgot", "side_effects", "quit", "other")
# ------------------------------------------------------------------------------------------- texts
class _Strict(BaseModel):
model_config = ConfigDict(extra="forbid", frozen=True)
class ReminderTexts(_Strict):
text: str
text_no_drug: str
taken: str
skip: str
side_effects: str
class SkipTexts(_Strict):
question: str
reasons: dict[str, str]
ack: dict[str, str]
class SideEffectTexts(_Strict):
question: str
options: dict[str, str]
questions: dict[str, str]
describe: str
ack: str
repeat: str
ask_dose: str
dose_yes: str
dose_no: str
class CheckinMedTexts(_Strict):
question: str
question_no_drug: str
yes: str
no: str
later: str
class QuestionTexts(_Strict):
prompt: str
no_prescription: str
class MedInfoTopic(_Strict):
section: str # med_qa section code (side_effects | onset | alcohol | ...)
question: str
lead: str = ""
fallback: str
notice: str = ""
class MedsTexts(_Strict):
status: str
reminder: ReminderTexts
taken_ack: str
stale: str
skip: SkipTexts
side_effects: SideEffectTexts
checkin: CheckinMedTexts
question: QuestionTexts
med_info: dict[str, MedInfoTopic]
def meds_texts_path() -> Path:
return get_settings().config_dir / "texts" / "meds.yaml"
@lru_cache
def load_meds_texts(path: Path | None = None) -> MedsTexts:
path = path or meds_texts_path()
return MedsTexts.model_validate(yaml.safe_load(path.read_text(encoding="utf-8")))
# ------------------------------------------------------------------------------------------- data helpers
class Rx(BaseModel):
"""Active prescription as the dialog needs it (drug INN + dose exactly as the doctor wrote them)."""
id: int | None
drug: str
dose: str
async def current_prescription(ctx: Ctx) -> Rx | None:
"""Latest prescription active today: clinical DB first, then the МИС integration (demo: in-memory)."""
today = ctx.now.date()
rows = await ctx.session.scalars(
select(PrescriptionRow).where(PrescriptionRow.token == ctx.token).order_by(PrescriptionRow.id.desc())
)
for row in rows:
if (row.start_date is None or row.start_date <= today) and (row.end_date is None or row.end_date >= today):
return Rx(id=row.id, drug=row.drug, dose=row.dose)
for p in await integrations.prescriptions.get_prescriptions(ctx.token):
if (p.start_date is None or p.start_date <= today) and (p.end_date is None or p.end_date >= today):
return Rx(id=None, drug=p.drug, dose=p.dose)
return None
async def checkin_of(ctx: Ctx, day: date) -> DailyCheckin | None:
return await ctx.session.scalar(
select(DailyCheckin).where(DailyCheckin.token == ctx.token, DailyCheckin.day == day)
)
async def save_checkin(ctx: Ctx, **fields: Any) -> DailyCheckin:
return await repo.upsert_checkin(ctx.session, token=ctx.token, day=ctx.now.date(), now=ctx.now, **fields)
def dose_ref(dose_id: int | None, day: date) -> str:
"""Button payload naming a dose: its row id, or the reminder day when there is no row."""
return str(dose_id) if dose_id is not None else day.isoformat()
async def todays_dose(ctx: Ctx) -> Dose | None:
"""Latest dose row planned for today (local day of `ctx.now`) — target of answers given without a button."""
rows = await ctx.session.scalars(select(Dose).where(Dose.token == ctx.token).order_by(Dose.id.desc()).limit(20))
today = ctx.now.date()
return next((r for r in rows if r.planned_at.astimezone(ctx.now.tzinfo).date() == today), None)
async def resolve_ref(ctx: Ctx, ref: str | None) -> tuple[bool, Dose | None]:
"""(fresh, dose row) for a button payload. Not fresh: missing/foreign/unknown ref or a dose of another day."""
if not ref:
return False, None
if ref.isdigit():
row = await ctx.session.get(Dose, int(ref))
if row is None or row.token != ctx.token:
return False, None
return row.planned_at.astimezone(ctx.now.tzinfo).date() == ctx.now.date(), row
try:
return date.fromisoformat(ref) == ctx.now.date(), None
except ValueError:
return False, None
async def _dose_row(ctx: Ctx, ref: str | None) -> Dose | None:
if ref is None:
return await todays_dose(ctx)
return (await resolve_ref(ctx, ref))[1]
async def record_dose(ctx: Ctx, taken: bool, *, status: str | None = None, ref: str | None = None) -> None:
"""Dose answer → `doses` row (`ref` from the button, else today's) + `daily_checkins.med_taken` (rules read
both)."""
row = await _dose_row(ctx, ref)
if row is not None:
row.status = status or ("taken" if taken else "skipped")
row.answered_at = ctx.now
await save_checkin(ctx, med_taken=taken)
async def record_side_effects(ctx: Ctx, codes: list[str], *, ref: str | None = None) -> None:
"""Merge side-effect codes into today's check-in (rule `side_effect_3d` counts days with any code)."""
today = await checkin_of(ctx, ctx.now.date())
merged = list(dict.fromkeys([*(today.side_effects if today and today.side_effects else []), *codes]))
await save_checkin(ctx, side_effects=merged)
row = await _dose_row(ctx, ref)
if row is not None:
row.side_effects = ",".join(c for c in dict.fromkeys([*(row.side_effects or "").split(","), *codes]) if c)
async def side_effect_reply(ctx: Ctx, codes: list[str]) -> Reply:
"""Reaction to newly reported side effects.
Same effect as yesterday → short support (the instruction was already shown; persistence over 3 days is the
rules' business). New effect with a known instruction question → the excerpt of the drug instruction
(med_qa, section side_effects) — never generated advice. Otherwise a plain acknowledgement.
"""
t = load_meds_texts().side_effects
yesterday = await checkin_of(ctx, ctx.now.date() - timedelta(days=1))
if yesterday is not None and set(codes) & set(yesterday.side_effects or []):
return Reply(text=t.repeat, kind="info")
rx = await current_prescription(ctx)
for code in codes:
if code in t.questions and rx is not None:
answer = await med_qa(rx.drug, scrub(t.questions[code]), topic="side_effects")
if answer is not None:
return Reply(text=answer.text, kind="info")
return Reply(text=t.ack, kind="info")
async def leave_to_idle(ctx: Ctx, event: Event, data: dict[str, Any] | None = None) -> Step:
"""Abandon the current sub-flow and let the idle state handle the event (menu button, system event...)."""
step = Step(next_state="idle", data=data or {})
idle = handler_for("idle")
return step.then(await idle(ctx, event)) if idle is not None else step
async def run_trigger_inline(ctx: Ctx, key: str, event: Event) -> Step | None:
"""Run an idle trigger from inside another flow (e.g. a reminder button pressed during the check-in)."""
handler = get_trigger(key)
return await handler(ctx, event) if handler is not None else None
# ------------------------------------------------------------------------------------------- reminder
def _reminder_buttons(ref: str) -> list[list[Button]]:
t = load_meds_texts().reminder
return [
[Button(label=t.taken, data=f"med:taken:{ref}")],
[Button(label=t.skip, data=f"med:skip:{ref}")],
[Button(label=t.side_effects, data=f"med:side_effects:{ref}")],
]
@trigger("system:med_reminder", replace=True)
async def _med_reminder(ctx: Ctx, event: Event) -> Step:
t = load_meds_texts().reminder
rx = await current_prescription(ctx)
drug = str(event.params.get("drug") or (rx.drug if rx else ""))
dose = str(event.params.get("dose") or (rx.dose if rx else ""))
try:
dose_id: int | None = int(event.params["dose_id"]) # type: ignore[call-overload]
except (KeyError, TypeError, ValueError):
dose_id = None
if dose_id is None and rx is not None and rx.id is not None:
# No scheduler row (demo console without #13): create the planned dose so the answer has a record.
row = Dose(token=ctx.token, prescription_id=rx.id, planned_at=ctx.now, status="pending")
ctx.session.add(row)
await ctx.session.flush()
dose_id = row.id
text = t.text.format(drug=drug, dose=dose).replace(" ", " ") if drug else t.text_no_drug
reply = Reply(text=text, buttons=_reminder_buttons(dose_ref(dose_id, ctx.now.date())), kind="question")
return Step(replies=[reply], data={"med_pending": None}) # key of earlier releases: the button carries it now
def _stale_reply() -> Step:
return Step(replies=[Reply(text=load_meds_texts().stale)])
async def _legacy(ctx: Ctx, event: Event) -> Step:
"""Payload-less medication button of an earlier release: the dose is unknown → treat as stale."""
return _stale_reply()
def _with_dose(fn: Any) -> Any:
"""Prefix-trigger handler wrapper: resolve the dose payload, answer stale buttons without recording."""
async def handler(ctx: Ctx, event: Event) -> Step:
ref = trigger_payload(event)
fresh, _row = await resolve_ref(ctx, ref)
if not fresh:
return _stale_reply()
return await fn(ctx, event, ref)
return handler
def med_trigger(name: str) -> Any:
"""Register `button:med:<name>:*` (with the dose payload) plus the legacy payload-less key as stale."""
def deco(fn: Any) -> Any:
trigger(f"button:med:{name}:*")(_with_dose(fn))
trigger(f"button:med:{name}")(_legacy)
return fn
return deco
@med_trigger("taken")
async def _taken(ctx: Ctx, event: Event, ref: str) -> Step:
await record_dose(ctx, True, ref=ref)
return Step(replies=[Reply(text=load_meds_texts().taken_ack)], notify_monitoring=True)
@med_trigger("skip")
async def _skip(ctx: Ctx, event: Event, ref: str) -> Step:
await record_dose(ctx, False, status="skipped", ref=ref)
t = load_meds_texts().skip
buttons = [[Button(label=t.reasons[r], data=f"med:skip_reason:{r}:{ref}")] for r in SKIP_REASONS]
return Step(replies=[Reply(text=t.question, buttons=buttons, kind="question")], notify_monitoring=True)
def _side_effect_question(ref: str) -> Reply:
t = load_meds_texts().side_effects
buttons = [[Button(label=t.options[c], data=f"med:se:{c}:{ref}")] for c in SIDE_EFFECT_CODES]
return Reply(text=t.question, buttons=buttons, kind="question")
def _skip_reason(reason: str) -> Any:
async def handler(ctx: Ctx, event: Event, ref: str) -> Step:
if reason == "side_effects":
return Step(replies=[_side_effect_question(ref)])
if reason == "quit":
# «Решил бросить» is the same signal as "брошу" in free text: rule `quit_intent` reacts (alert +
# delayed-onset explanation through the med_info action).
await save_checkin(ctx, quit_intent=True)
return Step(replies=[Reply(text=load_meds_texts().skip.ack["quit"])], notify_monitoring=True)
return Step(replies=[Reply(text=load_meds_texts().skip.ack[reason])])
return handler
for _reason in SKIP_REASONS:
med_trigger(f"skip_reason:{_reason}")(_skip_reason(_reason))
@med_trigger("side_effects")
async def _side_effects(ctx: Ctx, event: Event, ref: str) -> Step:
return Step(replies=[_side_effect_question(ref)])
async def _after_side_effects(ctx: Ctx, codes: list[str], ref: str | None) -> Step:
if ref is None:
row = await todays_dose(ctx)
ref = dose_ref(row.id if row is not None else None, ctx.now.date())
await record_side_effects(ctx, codes, ref=ref)
replies = [await side_effect_reply(ctx, codes)]
today = await checkin_of(ctx, ctx.now.date())
if (today is None or today.med_taken is None) and await current_prescription(ctx) is not None:
t = load_meds_texts().side_effects
buttons = [
[Button(label=t.dose_yes, data=f"med:dose:yes:{ref}")],
[Button(label=t.dose_no, data=f"med:dose:no:{ref}")],
]
replies.append(Reply(text=t.ask_dose, buttons=buttons, kind="question"))
return Step(replies=replies, next_state="idle", data={"med_se_ref": None}, notify_monitoring=True)
def _side_effect_code(code: str) -> Any:
async def handler(ctx: Ctx, event: Event, ref: str) -> Step:
if code == "other":
return Step(
replies=[Reply(text=load_meds_texts().side_effects.describe)],
next_state=SIDE_EFFECT_TEXT_STATE,
data={"med_se_ref": ref},
)
return await _after_side_effects(ctx, [code], ref)
return handler
for _code in SIDE_EFFECT_CODES:
med_trigger(f"se:{_code}")(_side_effect_code(_code))
@state(SIDE_EFFECT_TEXT_STATE)
async def _side_effect_text(ctx: Ctx, event: Event) -> Step:
if event.type != "text" or ctx.text is None:
return await leave_to_idle(ctx, event, {"med_se_ref": None})
facts = await ctx.gateway.extract_checkin(ctx.text)
# Store only normalised codes, never the description itself (it stays in the scrubbed message log).
codes = [str(c) for c in facts.side_effects] if facts is not None and facts.side_effects else ["other"]
ref = ctx.data.get("med_se_ref")
return await _after_side_effects(ctx, codes, str(ref) if ref else None)
def _dose_answer(taken: bool) -> Any:
async def handler(ctx: Ctx, event: Event, ref: str) -> Step:
await record_dose(ctx, taken, status=None if taken else "missed", ref=ref)
return Step(
replies=[Reply(text=load_meds_texts().taken_ack if taken else load_meds_texts().skip.ack["other"])],
notify_monitoring=True,
)
return handler
med_trigger("dose:yes")(_dose_answer(True))
med_trigger("dose:no")(_dose_answer(False))
# Free-text waiting states of this flow: a long-forgotten question must not swallow tomorrow's message.
abandonable("meds")
# ------------------------------------------------------------------------------------------- question menu
@trigger("button:menu:med_question", replace=True)
async def _menu_question(ctx: Ctx, event: Event) -> Step:
t = load_meds_texts().question
if await current_prescription(ctx) is None:
return Step(replies=[menu_reply(ctx, t.no_prescription)])
return Step(replies=[Reply(text=t.prompt)], next_state=QUESTION_STATE)
async def answer_question(ctx: Ctx, rx: Rx | None) -> str:
"""Answer from the instruction excerpt only (SPEC §7 п.5); dose questions and gaps → «уточните у врача»."""
if rx is None or ctx.text is None:
return load_meds_texts().question.no_prescription
answer = await med_qa(rx.drug, ctx.text)
return answer.text if answer is not None else fallback_text()
@state(QUESTION_STATE)
async def _question(ctx: Ctx, event: Event) -> Step:
if event.type != "text" or ctx.text is None:
return await leave_to_idle(ctx, event)
text = await answer_question(ctx, await current_prescription(ctx))
return Step(replies=[menu_reply(ctx, text)], next_state="idle")
# ------------------------------------------------------------------------------------------- rule actions
async def med_info_reply(ctx: Ctx, action: Action) -> Reply | None:
"""`med_info` action of a rule (quit_intent → onset, alcohol_3d → alcohol): instruction excerpt framed by
fixed texts. The rule already decided to say it; this only picks the words. None → engine template."""
spec = load_meds_texts().med_info.get(str(action.params.get("topic")))
if spec is None:
return None
rx = await current_prescription(ctx)
answer = await med_qa(rx.drug, scrub(spec.question), topic=spec.section) if rx is not None else None
body = answer.text if answer is not None else spec.fallback
return Reply(text="\n\n".join(p for p in (spec.lead, body, spec.notice) if p), kind="info")
register_med_info_handler(med_info_reply)
+465
View File
@@ -0,0 +1,465 @@
"""Daily check-in and scheduled-scale intros (#37, ADR-002, CLINICAL_MODULES §1–2, case «мониторинг после
обращения»).
Check-in = core (mood 0–10, dose taken) + the module question (veteran: sleep hours + nightmares; alcohol for
veterans and for anyone with a positive AUDIT-C) + free text:
system:daily_checkin / menu «Как я сегодня»
──▶ daily.mood buttons 0–10 (a bare number is accepted too)
──▶ daily.med only with a prescription and no dose answer today (the reminder may have got it)
──▶ daily.sleep <4 / 4–6 / 6–8 / >8 h ┐ module veteran
──▶ daily.nightmares yes / no ┘
──▶ daily.alcohol none / 1–2 / more (veteran or AUDIT-C positive)
──▶ daily.text «Как вы? Пару слов» (skip button) → extract_checkin on the SCRUBBED text
──▶ [daily.alcohol_amount] alcohol mentioned in the text but not asked by button → amount buttons
──▶ idle upsert done, `notify_monitoring` → rules (lane B) → dialog actions
Data algorithm (medical data, jury requirement):
* Every button answer is written to `daily_checkins` immediately (`repo.upsert_checkin` merges non-None
fields), so an interrupted check-in (crisis, the patient walking away) still leaves today's mood on record —
the "mood ≤ 3 three days in a row" rule must not miss a day because the free text was never written.
* Button answers are the patient's own and always win over the LLM: from the free text the LLM may only add
side effects, the quit intent and fields that were not asked by buttons today.
* The LLM never decides anything here: alerts and follow-ups (delayed-onset explanation, unscheduled PHQ-9,
alcohol interaction, visit offer) come from the rules through `MonitoringService` (ADR-007).
* A free-text message classified POSSIBLE by the safety layer is taken over by the engine (C-SSRS). When the
screen returns to `daily.text` the message counts as the answer: the check-in is closed, not asked again.
* A check-in left unfinished is dropped on the next event of a later day, or after
`settings.abandon_after_hours` of silence (engine abandon timeout, #56) — answers given so far stay stored.
Also here: friendly intros for scheduled scales (pcl5, phq9, dar5, auditc), registered with
`register_scale_intro` so the engine's `scale_due` / rule-triggered sessions show them.
"""
import re
from datetime import datetime
from functools import lru_cache
from pathlib import Path
from typing import Any
import yaml
from pydantic import BaseModel, ConfigDict
from app.config import get_settings
from app.core.intake_runtime import intake_enabled
from app.core.rules import load_rules
from app.db import repo
from app.dialog.contracts import Button, Event, Reply
from app.dialog.flows import intake
from app.dialog.flows.core import menu_reply
from app.dialog.flows.meds import (
answer_question,
checkin_of,
current_prescription,
leave_to_idle,
load_meds_texts,
record_dose,
record_side_effects,
run_trigger_inline,
save_checkin,
side_effect_reply,
)
from app.dialog.fsm import Ctx, Step, abandonable, on_enter, state, trigger
from app.dialog.scale_session import register_scale_intro
PREFIX = "daily"
STEPS = ("mood", "med", "sleep", "nightmares", "alcohol", "text")
ALCOHOL_AMOUNT = "alcohol_amount"
SLEEP_BUCKETS = ("<4", "4-6", "6-8", ">8")
ALCOHOL_BUCKETS = ("none", "1-2", "more")
_NUMBER = re.compile(r"^\s*(\d{1,2})\s*$")
# ------------------------------------------------------------------------------------------- texts
class _Strict(BaseModel):
model_config = ConfigDict(extra="forbid", frozen=True)
class DailySettings(_Strict):
sleep_modules: list[str]
alcohol_daily_modules: list[str]
intro_scales: list[str]
class Question(_Strict):
question: str
class OptionsQuestion(_Strict):
question: str
options: dict[str, str]
class YesNoQuestion(_Strict):
question: str
yes: str
no: str
class AlcoholQuestion(_Strict):
question: str
clarify: str
options: dict[str, str]
class TextQuestion(_Strict):
question: str
skip: str
class CheckinTexts(_Strict):
start: str
mood: Question
sleep: OptionsQuestion
nightmares: YesNoQuestion
alcohol: AlcoholQuestion
text: TextQuestion
buttons_hint: str
done: str
already_done: str
class VisitTexts(_Strict):
requested: str
class ScaleIntros(_Strict):
scheduled: dict[str, str]
unscheduled: dict[str, str]
class DailyTexts(_Strict):
status: str
settings: DailySettings
checkin: CheckinTexts
visit: VisitTexts
scale_intros: ScaleIntros
def daily_texts_path() -> Path:
return get_settings().config_dir / "texts" / "veteran_daily.yaml"
@lru_cache
def load_daily_texts(path: Path | None = None) -> DailyTexts:
path = path or daily_texts_path()
return DailyTexts.model_validate(yaml.safe_load(path.read_text(encoding="utf-8")))
# ------------------------------------------------------------------------------------------- step logic
def _today(ctx: Ctx) -> str:
return ctx.now.date().isoformat()
def _ck(ctx: Ctx) -> dict[str, Any]:
return dict(ctx.data.get("checkin") or {})
async def _modules(ctx: Ctx) -> set[str]:
profile = await ctx.profile()
return set(profile.modules or []) if profile is not None else set()
async def _auditc_positive(ctx: Ctx) -> bool:
"""Latest AUDIT-C at or above the sex-specific cut-off of rule `auditc_positive` (one threshold, one place)."""
results = await repo.list_scale_results(ctx.session, ctx.token, "auditc")
if not results:
return False
rule = next((r for r in load_rules().rules if r.id == "auditc_positive"), None)
if rule is None:
return False
profile = await ctx.profile()
key = "threshold_f" if profile is not None and profile.sex == "f" else "threshold_m"
return results[-1].total >= int(rule.params[key])
async def _applicable(ctx: Ctx, step: str) -> bool:
settings = load_daily_texts().settings
if step in ("mood", "text"):
return True
if step == "med":
today = await checkin_of(ctx, ctx.now.date())
return (today is None or today.med_taken is None) and await current_prescription(ctx) is not None
modules = await _modules(ctx)
if step in ("sleep", "nightmares"):
return bool(modules & set(settings.sleep_modules))
if step == "alcohol":
return bool(modules & set(settings.alcohol_daily_modules)) or await _auditc_positive(ctx)
return False
async def _next_step(ctx: Ctx, after: str | None) -> str:
rest = STEPS if after is None else STEPS[STEPS.index(after) + 1 :]
for step in rest:
if await _applicable(ctx, step):
return step
return "text"
def _buttons(step: str, options: list[tuple[str, str]]) -> list[list[Button]]:
return [[Button(label=label, data=f"{PREFIX}:{step}:{value}")] for value, label in options]
async def _question(ctx: Ctx, step: str, note: str = "") -> Reply:
t = load_daily_texts().checkin
buttons: list[list[Button]]
if step == "mood":
text = t.mood.question
# Two rows (0–5, 6–10): 11 buttons in one column are hard to use on a phone.
buttons = [
[Button(label=str(v), data=f"{PREFIX}:mood:{v}") for v in range(6)],
[Button(label=str(v), data=f"{PREFIX}:mood:{v}") for v in range(6, 11)],
]
elif step == "med":
m = load_meds_texts().checkin
rx = await current_prescription(ctx)
text = m.question.format(drug=rx.drug) if rx is not None else m.question_no_drug
buttons = _buttons(step, [("yes", m.yes), ("no", m.no), ("later", m.later)])
elif step == "sleep":
text = t.sleep.question
buttons = _buttons(step, [(b, t.sleep.options[b]) for b in SLEEP_BUCKETS])
elif step == "nightmares":
text = t.nightmares.question
buttons = _buttons(step, [("yes", t.nightmares.yes), ("no", t.nightmares.no)])
elif step in ("alcohol", ALCOHOL_AMOUNT):
text = t.alcohol.question if step == "alcohol" else t.alcohol.clarify
buttons = _buttons(step, [(b, t.alcohol.options[b]) for b in ALCOHOL_BUCKETS])
else:
text = t.text.question
buttons = _buttons("text", [("skip", t.text.skip)])
return Reply(text="\n\n".join(p for p in (note, text) if p), buttons=buttons, kind="question")
async def start_checkin(ctx: Ctx, *, from_menu: bool) -> Step:
if ctx.data.get("checkin_done") == _today(ctx):
# The scheduler may fire twice; the menu button gets a friendly note instead of a second check-in.
return Step(replies=[menu_reply(ctx, load_daily_texts().checkin.already_done)]) if from_menu else Step()
first = await _next_step(ctx, None)
ck = {"day": _today(ctx), "answered": [], "started_at": ctx.now.isoformat()}
return Step(
replies=[Reply(text=load_daily_texts().checkin.start)],
next_state=f"{PREFIX}.{first}",
data={"checkin": ck},
)
def _finish(ctx: Ctx, replies: list[Reply] | None = None) -> Step:
done = Reply(text=load_daily_texts().checkin.done)
return Step(
replies=[done, *(replies or [])],
next_state="idle",
data={"checkin": None, "checkin_done": _today(ctx)},
notify_monitoring=True,
)
def _consumed_by_safety(ctx: Ctx, ck: dict[str, Any]) -> bool:
"""The free-text answer was taken over by the safety layer (C-SSRS screen) after the question was asked."""
asked = ck.get("text_asked_at")
if not asked:
return False
for event in reversed(ctx.data.get("safety_events") or []):
if event.get("state") == f"{PREFIX}.text" and datetime.fromisoformat(event["ts"]) >= datetime.fromisoformat(
asked
):
return True
return False
@on_enter(PREFIX)
async def _enter(ctx: Ctx) -> Step:
step = ctx.state.split(".", 1)[1] if "." in ctx.state else ""
ck = _ck(ctx)
if not ck or ck.get("day") != _today(ctx):
return Step(next_state="idle", data={"checkin": None})
if step == "text":
if _consumed_by_safety(ctx, ck):
return _finish(ctx)
return Step(
replies=[await _question(ctx, step)], data={"checkin": {**ck, "text_asked_at": ctx.now.isoformat()}}
)
return Step(replies=[await _question(ctx, step)])
async def _advance(ctx: Ctx, step: str, ck: dict[str, Any]) -> Step:
ck = {**ck, "answered": [*ck.get("answered", []), step]}
if step == ALCOHOL_AMOUNT:
return _finish(ctx)
nxt = await _next_step(ctx, step)
return Step(next_state=f"{PREFIX}.{nxt}", data={"checkin": ck})
async def _record_button(ctx: Ctx, step: str, value: str) -> bool:
"""Store one button answer right away. False = value not valid for this step (stale/forged button)."""
if step == "mood" and value.isdigit() and 0 <= int(value) <= 10:
await save_checkin(ctx, mood=int(value))
elif step == "med" and value in ("yes", "no", "later"):
if value != "later": # «Ещё нет» is not a missed dose: nothing is recorded yet
await record_dose(ctx, value == "yes", status=None if value == "yes" else "missed")
elif step == "sleep" and value in SLEEP_BUCKETS:
await save_checkin(ctx, sleep_bucket=value)
elif step == "nightmares" and value in ("yes", "no"):
await save_checkin(ctx, nightmares=value == "yes")
elif step in ("alcohol", ALCOHOL_AMOUNT) and value in ALCOHOL_BUCKETS:
await save_checkin(ctx, alcohol_bucket=value)
else:
return False
return True
async def _free_text(ctx: Ctx, ck: dict[str, Any]) -> Step:
assert ctx.text is not None
facts = await ctx.gateway.extract_checkin(ctx.text)
answered = set(ck.get("answered", []))
fields: dict[str, Any] = {"free_text_scrubbed": ctx.text}
replies: list[Reply] = []
clarify_alcohol = False
if facts is not None:
if facts.quit_intent:
fields["quit_intent"] = True
# Buttons win: the LLM only fills what was not asked today.
if "med" not in answered and facts.med_taken is not None:
today = await checkin_of(ctx, ctx.now.date())
if today is None or today.med_taken is None:
fields["med_taken"] = facts.med_taken
if "sleep" not in answered and facts.sleep_hours_bucket is not None:
fields["sleep_bucket"] = facts.sleep_hours_bucket
if "nightmares" not in answered and facts.nightmares is not None:
fields["nightmares"] = facts.nightmares
if "alcohol" not in answered and facts.alcohol is not None:
fields["alcohol_bucket"] = facts.alcohol # provisional; the amount buttons below refine it
clarify_alcohol = facts.alcohol != "none"
if facts.side_effects:
await record_side_effects(ctx, [str(c) for c in facts.side_effects])
# One topic per message: when the text also carries a quit intent or alcohol, the rules answer
# those (delayed onset / interaction); the side-effect excerpt would only bury that message.
if not facts.quit_intent and facts.alcohol in (None, "none"):
replies.append(await side_effect_reply(ctx, [str(c) for c in facts.side_effects]))
await save_checkin(ctx, **fields)
if clarify_alcohol:
return Step(replies=replies, next_state=f"{PREFIX}.{ALCOHOL_AMOUNT}", data={"checkin": ck})
return _finish(ctx, replies)
abandonable(PREFIX)
@state(PREFIX)
async def _handle(ctx: Ctx, event: Event) -> Step:
step = ctx.state.split(".", 1)[1] if "." in ctx.state else ""
ck = _ck(ctx)
if not ck or ck.get("day") != _today(ctx):
# Unfinished check-in from an earlier day: drop it and handle the event as idle would.
return await leave_to_idle(ctx, event, {"checkin": None})
if event.type == "button" and event.button:
parts = event.button.split(":")
if len(parts) == 3 and parts[0] == PREFIX and parts[1] == step:
if step == "text" and parts[2] == "skip":
return _finish(ctx)
if await _record_button(ctx, step, parts[2]):
return await _advance(ctx, step, ck)
if parts[0] == "med":
# A reminder button pressed mid check-in: answer it, then repeat the current question.
inline = await run_trigger_inline(ctx, f"button:{event.button}", event)
if inline is not None:
return inline if inline.next_state else inline.then(Step(replies=[await _question(ctx, step)]))
if parts[0] in (PREFIX, "scale"):
return Step(replies=[await _question(ctx, step)]) # stale button of an earlier question
return await leave_to_idle(ctx, event, {"checkin": None})
if event.type == "system" and event.system:
if event.system == "daily_checkin":
return Step(replies=[await _question(ctx, step)])
inline = await run_trigger_inline(ctx, f"system:{event.system}", event)
if inline is None:
return Step()
return inline if inline.next_state else inline.then(Step(replies=[await _question(ctx, step)]))
if event.type == "text" and ctx.text is not None:
if step == "text":
return await _free_text(ctx, ck)
number = _NUMBER.match(ctx.text)
if step == "mood" and number and await _record_button(ctx, "mood", number.group(1)):
return await _advance(ctx, step, ck)
return Step(replies=[await _question(ctx, step, note=load_daily_texts().checkin.buttons_hint)])
return Step(replies=[await _question(ctx, step)])
# ------------------------------------------------------------------------------------------- idle triggers
@trigger("system:daily_checkin", replace=True)
async def _daily_checkin(ctx: Ctx, event: Event) -> Step:
if intake_enabled(): # the intake diary (#53, feature flag) replaces this check-in when released
return await intake.checkin(ctx, event)
return await start_checkin(ctx, from_menu=False)
@trigger("button:menu:checkin", replace=True)
async def _menu_checkin(ctx: Ctx, event: Event) -> Step:
if intake_enabled():
return await intake.checkin(ctx, event)
return await start_checkin(ctx, from_menu=True)
@trigger("button:menu:visit", replace=True)
async def _visit(ctx: Ctx, event: Event) -> Step:
# «Записаться раньше» (rule action suggest_visit): no online booking yet — the request is a fact for lane B
# (#38) and the dashboard, like `contact_requests`. No alert is created here (ADR-007).
requests = [*(ctx.data.get("visit_requests") or []), {"ts": ctx.now.isoformat()}][-20:]
return Step(
replies=[Reply(text=load_daily_texts().visit.requested)],
data={"visit_requests": requests},
notify_monitoring=True,
)
@trigger("text:idle", replace=True)
async def _idle_text(ctx: Ctx, event: Event) -> Step:
"""Free text outside the check-in: drug question → instruction excerpt; side effects / quit intent / alcohol
→ today's check-in (the rules react); otherwise the neutral support phrase with the menu."""
assert ctx.text is not None
facts = await ctx.gateway.extract_checkin(ctx.text)
if facts is None:
return Step(replies=[menu_reply(ctx, await ctx.gateway.phrase("support"))])
fields: dict[str, Any] = {}
if facts.quit_intent:
fields["quit_intent"] = True
if facts.alcohol is not None and facts.alcohol != "none":
today = await checkin_of(ctx, ctx.now.date())
if today is None or today.alcohol_bucket in (None, "none"):
fields["alcohol_bucket"] = facts.alcohol
text: str | None = None
if facts.side_effects:
codes = [str(c) for c in facts.side_effects]
await record_side_effects(ctx, codes)
if not facts.quit_intent:
text = (await side_effect_reply(ctx, codes)).text
if fields:
await save_checkin(ctx, **fields)
if text is None and facts.med_question and not facts.quit_intent:
text = await answer_question(ctx, await current_prescription(ctx))
if text is None:
text = await ctx.gateway.phrase("support")
return Step(replies=[menu_reply(ctx, text)], notify_monitoring=bool(fields or facts.side_effects))
# ------------------------------------------------------------------------------------------- scale intros
def _scale_intro(ctx: Ctx, scale_id: str, scheduled: bool, return_state: str) -> str | None:
"""Friendly intro of a scheduled (or rule-triggered) scale. Onboarding sessions (#36) have their own texts."""
if ".onboarding" in return_state:
return None
intros = load_daily_texts().scale_intros
return (intros.scheduled if scheduled else intros.unscheduled).get(scale_id)
for _scale_id in load_daily_texts().settings.intro_scales:
register_scale_intro(_scale_id, _scale_intro)
+458
View File
@@ -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:<question>:<choice>"
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)
+57 -4
View File
@@ -22,6 +22,14 @@ Name resolution: an exact registration wins, otherwise the longest registered do
`@trigger("system:daily_checkin")`, `@trigger("text:idle")`) let flows take over what the core `idle` state does `@trigger("system:daily_checkin")`, `@trigger("text:idle")`) let flows take over what the core `idle` state does
with menu buttons, system events and free text; the core registers defaults, a flow overrides with with menu buttons, system events and free text; the core registers defaults, a flow overrides with
`replace=True`. `replace=True`.
Prefix triggers carry data in the button (#56): `@trigger("button:med:taken:*")` handles every button whose data
starts with "med:taken:"; the rest of the data ("med:taken:42" → "42") reaches the handler as
`trigger_payload(event)`. An exact key always wins over a prefix, a longer prefix over a shorter one.
Abandoned steps (#56): a flow marks its question states with `abandonable("daily")`; if the patient stays silent
in such a state longer than `settings.abandon_after_hours`, the engine returns the FSM to `idle` before
handling the next event (answers already stored stay stored).
""" """
from __future__ import annotations from __future__ import annotations
@@ -103,6 +111,11 @@ EnterHook = Callable[[Ctx], Awaitable[Step]]
_handlers: dict[str, Handler] = {} _handlers: dict[str, Handler] = {}
_enter_hooks: dict[str, EnterHook] = {} _enter_hooks: dict[str, EnterHook] = {}
_triggers: dict[str, Handler] = {} _triggers: dict[str, Handler] = {}
_prefix_triggers: dict[str, Handler] = {} # "button:med:taken:" → handler (registered as "button:med:taken:*")
_abandonable: dict[str, Callable[[Ctx], bool] | None] = {}
PREFIX_WILDCARD = "*"
PAYLOAD_PARAM = "trigger_payload"
def state(name: str) -> Callable[[Handler], Handler]: def state(name: str) -> Callable[[Handler], Handler]:
@@ -133,18 +146,45 @@ def trigger(key: str, *, replace: bool = False) -> Callable[[Handler], Handler]:
"""Register a handler for an idle-menu button / system event / idle free text. """Register a handler for an idle-menu button / system event / idle free text.
Keys: "button:<Button.data>", "system:<SystemEventKind>", "text:idle". The core registers defaults; Keys: "button:<Button.data>", "system:<SystemEventKind>", "text:idle". The core registers defaults;
flows pass `replace=True` to take one over. flows pass `replace=True` to take one over. A key ending in ":*" is a prefix trigger (see module doc).
""" """
table, name = _triggers, key
if key.endswith(":" + PREFIX_WILDCARD):
table, name = _prefix_triggers, key[: -len(PREFIX_WILDCARD)]
def deco(fn: Handler) -> Handler: def deco(fn: Handler) -> Handler:
if key in _triggers and not replace: if name in table and not replace:
raise ValueError(f"trigger '{key}' already registered (use replace=True to override)") raise ValueError(f"trigger '{key}' already registered (use replace=True to override)")
_triggers[key] = fn table[name] = fn
return fn return fn
return deco return deco
def trigger_payload(event: Event) -> str | None:
"""Data after the matched prefix of a prefix trigger ("med:taken:42" → "42"); None for exact triggers."""
value = event.params.get(PAYLOAD_PARAM)
return str(value) if value is not None else None
def abandonable(prefix: str, when: Callable[[Ctx], bool] | None = None) -> None:
"""Mark states `prefix` / `prefix.*` as question steps the engine may abandon after a long silence.
`when(ctx)` may veto it for a particular situation (e.g. a scale that is part of an onboarding).
"""
if prefix in _abandonable:
raise ValueError(f"abandonable '{prefix}' already registered")
_abandonable[prefix] = when
def is_abandonable(ctx: Ctx) -> bool:
key = _resolve(ctx.state, _abandonable)
if key is None:
return False
when = _abandonable[key]
return when is None or when(ctx)
def _resolve(name: str, table: dict[str, Any]) -> str | None: def _resolve(name: str, table: dict[str, Any]) -> str | None:
candidate = name candidate = name
while candidate: while candidate:
@@ -167,7 +207,20 @@ def enter_hook_for(name: str) -> EnterHook | None:
def get_trigger(key: str) -> Handler | None: def get_trigger(key: str) -> Handler | None:
return _triggers.get(key) """Exact trigger, else the longest matching prefix trigger (wrapped so it sees the payload)."""
exact = _triggers.get(key)
if exact is not None:
return exact
matches = [p for p in _prefix_triggers if key.startswith(p) and len(key) > len(p)]
if not matches:
return None
prefix = max(matches, key=len)
fn, payload = _prefix_triggers[prefix], key[len(prefix) :]
async def with_payload(ctx: Ctx, event: Event) -> Step:
return await fn(ctx, event.model_copy(update={"params": {**event.params, PAYLOAD_PARAM: payload}}))
return with_payload
def is_registered(name: str) -> bool: def is_registered(name: str) -> bool:
+57 -5
View File
@@ -15,13 +15,22 @@ Algorithm per item:
4. Every answer is stored immediately (`item_responses`), so a session interrupted by a crisis keeps its data. 4. Every answer is stored immediately (`item_responses`), so a session interrupted by a crisis keeps its data.
5. After each answer the scale is scored deterministically (`app.core.scales.score`). If a crisis flag from 5. After each answer the scale is scored deterministically (`app.core.scales.score`). If a crisis flag from
`config/texts/dialog.yaml: settings.crisis_flags` fires (PHQ-9 item 9 ≥ 1, C-SSRS high) the session ends at `config/texts/dialog.yaml: settings.crisis_flags` fires (PHQ-9 item 9 ≥ 1, C-SSRS high) the session ends at
once: partial result saved, fixed crisis script, C-SSRS queued (unless this *is* the C-SSRS). A positive once: partial result saved, fixed crisis script, C-SSRS queued (unless this *is* the C-SSRS, or a C-SSRS was
item 9 is acted upon even in a half-finished questionnaire. completed less than `settings.cssrs_cooldown_hours` ago — the patient is not asked the same suicide screen
twice in a day; the crisis script and the alert still happen). A positive item 9 is acted upon even in a
half-finished questionnaire.
6. On completion: `scale_results` row (total, severity, item 9, scheduled flag, flags) → the engine commits 6. On completion: `scale_results` row (total, severity, item 9, scheduled flag, flags) → the engine commits
and calls `MonitoringService.after_data_change()` (rules decide alerts — ADR-007) → back to `return_state`. and calls `MonitoringService.after_data_change()` (rules decide alerts — ADR-007) → back to `return_state`.
The patient never sees a score or a severity label: that would read as a diagnosis. The patient never sees a score or a severity label: that would read as a diagnosis.
Intros (#56): `start_scale(intro=None)` looks the intro up in a registry filled by flows
(`register_scale_intro("phq9", fn_or_text)`), so scales started by the engine (`scale_due`, rule actions) get
the flow's friendly intro without the flow hooking `scale.<id>` itself.
""" """
from collections.abc import Callable
from dataclasses import replace
from datetime import timedelta
from typing import Any from typing import Any
from app.core.privacy import scrub from app.core.privacy import scrub
@@ -29,10 +38,28 @@ from app.core.scales import Item, ScaleDef, ScaleResult, load_scale, score
from app.db import repo from app.db import repo
from app.dialog.contracts import Button, Event, Reply from app.dialog.contracts import Button, Event, Reply
from app.dialog.crisis import crisis_open, open_crisis from app.dialog.crisis import crisis_open, open_crisis
from app.dialog.fsm import Ctx, Step, on_enter, state from app.dialog.fsm import RESTING_STATES, Ctx, Step, abandonable, is_abandonable, on_enter, state
PREFIX = "scale" PREFIX = "scale"
# (ctx, scale_id, scheduled, return_state) → intro text or None (no intro in this situation).
IntroFn = Callable[[Ctx, str, bool, str], str | None]
_intros: dict[str, str | IntroFn] = {}
def register_scale_intro(scale_id: str, intro: str | IntroFn, *, replace: bool = False) -> None:
"""Register the intro shown before `scale_id` when a session is started without an explicit `intro`."""
if scale_id in _intros and not replace:
raise ValueError(f"intro of scale '{scale_id}' already registered (use replace=True to override)")
_intros[scale_id] = intro
def scale_intro(ctx: Ctx, scale_id: str, *, scheduled: bool, return_state: str) -> str | None:
intro = _intros.get(scale_id)
if intro is None or isinstance(intro, str):
return intro or None
return intro(ctx, scale_id, scheduled, return_state) or None
def scale_state(scale_id: str) -> str: def scale_state(scale_id: str) -> str:
return f"{PREFIX}.{scale_id}" return f"{PREFIX}.{scale_id}"
@@ -56,9 +83,11 @@ async def start_scale(
"""Begin a session of `scale_id`; returns a Step moving the FSM to `scale.<id>`. """Begin a session of `scale_id`; returns a Step moving the FSM to `scale.<id>`.
`scheduled=False` marks an unscheduled administration (triggered by a rule or a safety signal). `scheduled=False` marks an unscheduled administration (triggered by a rule or a safety signal).
`intro` — optional extra text shown before the scale instruction. `intro` — optional extra text shown before the scale instruction; None → the registered intro, if any.
""" """
scale = load_scale(scale_id) # fail fast on an unknown id scale = load_scale(scale_id) # fail fast on an unknown id
if intro is None:
intro = scale_intro(ctx, scale_id, scheduled=scheduled, return_state=return_state)
sessions = dict(ctx.data.get("scales") or {}) sessions = dict(ctx.data.get("scales") or {})
sessions[scale_id] = { sessions[scale_id] = {
"idx": 0, "idx": 0,
@@ -98,6 +127,28 @@ def _session(ctx: Ctx) -> tuple[ScaleDef, dict[str, Any]] | None:
return load_scale(scale_id), session return load_scale(scale_id), session
def _abandon_allowed(ctx: Ctx) -> bool:
"""A silent scale session may be abandoned only if what it returns to may be (idle, a check-in step) —
never an onboarding chain, which would otherwise be skipped."""
found = _session(ctx)
if found is None:
return True
back = str(found[1].get("return_state") or "idle")
if back in RESTING_STATES:
return True
return back != ctx.state and is_abandonable(replace(ctx, state=back))
abandonable(PREFIX, _abandon_allowed)
async def recent_risk_screen(ctx: Ctx, screen: str) -> bool:
"""A complete risk screen (C-SSRS) less than `cssrs_cooldown_hours` ago (scale_results, any trigger)."""
since = ctx.now - timedelta(hours=ctx.texts.settings.cssrs_cooldown_hours)
results = await repo.list_scale_results(ctx.session, ctx.token, screen)
return any(r.complete and r.ts >= since for r in results)
@on_enter(PREFIX) @on_enter(PREFIX)
async def _enter(ctx: Ctx) -> Step: async def _enter(ctx: Ctx) -> Step:
found = _session(ctx) found = _session(ctx)
@@ -231,6 +282,7 @@ async def _finish(
else: else:
crisis = open_crisis(ctx, categories=[], source=f"scale:{scale.id}") crisis = open_crisis(ctx, categories=[], source=f"scale:{scale.id}")
step = done.then(crisis) step = done.then(crisis)
if scale.id != risk_screen: # Same-day repeat guard: a C-SSRS completed within the cooldown already answered the screening question.
if scale.id != risk_screen and not await recent_risk_screen(ctx, risk_screen):
step.data["pending_scales"] = add_pending(ctx.data, risk_screen, scheduled=False) step.data["pending_scales"] = add_pending(ctx.data, risk_screen, scheduled=False)
return step return step
+1
View File
@@ -19,6 +19,7 @@ class DialogSettings(_Strict):
max_safety_events: int = 50 max_safety_events: int = 50
crisis_flags: list[str] = Field(default_factory=list) crisis_flags: list[str] = Field(default_factory=list)
risk_screen_scale: str = "cssrs_screen" risk_screen_scale: str = "cssrs_screen"
abandon_after_hours: float = 6
class Commands(_Strict): class Commands(_Strict):
+6
View File
@@ -12,6 +12,7 @@ from datetime import datetime
from zoneinfo import ZoneInfo from zoneinfo import ZoneInfo
from sqlalchemy import select, text from sqlalchemy import select, text
from sqlalchemy.exc import SQLAlchemyError
from app.core import clock from app.core import clock
from app.core.intake_runtime import due_scales, intake_enabled, load_runtime from app.core.intake_runtime import due_scales, intake_enabled, load_runtime
@@ -21,6 +22,7 @@ from app.db.intake import QuestionnaireTask
from app.db.session import get_clinical_session from app.db.session import get_clinical_session
from app.dialog.contracts import Event, Reply, get_engine from app.dialog.contracts import Event, Reply, get_engine
from app.integrations.base import integrations from app.integrations.base import integrations
from app.services import monitoring
from app.services import questionnaire as service from app.services import questionnaire as service
log = logging.getLogger(__name__) log = logging.getLogger(__name__)
@@ -163,6 +165,10 @@ async def _run():
async def start(): async def start():
global _worker global _worker
try:
await monitoring.init_clock_persistence()
except (OSError, SQLAlchemyError) as exc:
log.error("clock persistence unavailable error=%s", type(exc).__name__)
if _worker is None and intake_enabled(): if _worker is None and intake_enabled():
_worker = asyncio.create_task(_run()) _worker = asyncio.create_task(_run())
+157
View File
@@ -0,0 +1,157 @@
"""Monitoring service (#38): patient snapshot → deterministic rules → persisted alerts (ADR-007).
`after_data_change(token)` is called by the dialog engine after it commits clinical data (and by the scheduler
on silence). It builds a `PatientSnapshot`, runs `app.core.rules.evaluate`, then in one transaction:
* writes every `alert`/`crisis` action to `alerts` (a crisis is an `acute` alert) and, for a crisis,
an audit record `crisis_acute` (guardrail #6);
* applies `enhanced_monitoring` to the profile;
* refreshes the cached `status_color` of the profile (ADR-006; the source of truth stays the open alerts).
After the commit the doctor is notified through `integrations.notifier` (a crisis also goes to the doctor on
duty), and only the *dialog* actions are returned to the engine — it never creates alerts itself.
A crisis rule whose acute alert is still open stays silent: the crisis script is already running and the
doctor already knows, so it is neither restarted nor re-alerted (`rules.evaluate` never suppresses crises).
The virtual clock offset is persisted in `clock_state` (ADR-005) by `init_clock_persistence()`.
"""
import asyncio
import logging
from datetime import datetime, timedelta
from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy.ext.asyncio import AsyncSession
from app.core import clock
from app.core.rules import Action, evaluate, patient_status
from app.core.types import AlertLevel
from app.db import repo
from app.db.audit import audit
from app.db.session import get_clinical_session
from app.integrations.base import integrations
from app.services.contracts import register_monitoring
from app.services.snapshot import build_snapshot
log = logging.getLogger(__name__)
SYSTEM_ACTOR = "system"
# Notification address of the doctor on duty: acute alerts must reach someone even when the attending doctor
# is off (SPEC: crisis → immediate escalation). The notifier maps it to the duty channel.
ON_DUTY_DOCTOR = "on-duty"
# Persisted by the service itself, never passed to the dialog engine.
_SERVICE_ACTIONS = frozenset({"alert", "enhanced_monitoring"})
async def refresh_status(session: AsyncSession, token: str) -> str:
"""Recompute and cache the dashboard colour from open alerts (also call it after an ack)."""
alerts = await repo.list_alerts(session, token=token, open_only=True)
status = patient_status(AlertLevel[a.level.upper()] for a in alerts)
await repo.upsert_profile(session, token=token, status_color=status.color)
return status.color
def _drop_running_crises(actions: list[Action], open_acute: set[str]) -> list[Action]:
"""Silence every action of a crisis rule whose acute alert is still open (see module docstring)."""
running = {a.code for a in actions if a.type == "crisis" and a.code in open_acute}
return [a for a in actions if a.code not in running]
class RuleMonitoring:
async def after_data_change(self, token: str) -> list[Action]:
now = clock.now()
notify: list[tuple[str, str, str]] = [] # (doctor_id, level, text) — sent after the commit
async with get_clinical_session() as s:
profile = await repo.get_profile(s, token)
snapshot = await build_snapshot(s, token, now, profile=profile)
if profile is None or snapshot is None:
return []
open_acute = {a.code for a in await repo.list_alerts(s, token=token, open_only=True) if a.level == "acute"}
actions = _drop_running_crises(evaluate(snapshot), open_acute)
for action in actions:
if action.type == "enhanced_monitoring":
profile.enhanced_monitoring = True
if not action.creates_alert or action.level is None:
continue
alert = await repo.add_alert(
s,
token=token,
level=action.level,
code=action.code,
reason=action.reason_ru,
ts=now,
params=action.params,
)
level = action.level.name.lower()
notify.append((profile.doctor_id, level, action.reason_ru))
if action.type == "crisis":
await audit(
s,
actor=SYSTEM_ACTOR,
action="crisis_acute",
token=token,
details={"alert_id": alert.id, "code": action.code},
ts=now,
)
notify.append((ON_DUTY_DOCTOR, level, action.reason_ru))
await refresh_status(s, token)
log.info("monitoring token=%s actions=%s", token, [a.code for a in actions])
for doctor_id, level, text in notify:
await _notify(doctor_id, level, text, token)
return [a for a in actions if a.type not in _SERVICE_ACTIONS]
async def is_in_crisis(self, token: str) -> bool:
async with get_clinical_session() as s:
alerts = await repo.list_alerts(s, token=token, open_only=True)
return any(a.level == "acute" for a in alerts)
async def _notify(doctor_id: str, level: str, text: str, token: str) -> None:
# The alert is already committed and visible on the dashboard; a failing channel must not undo it.
try:
await integrations.notifier.send(doctor_id, level, text, token) # type: ignore[arg-type]
except Exception as exc: # noqa: BLE001 — any channel error
log.warning("notify failed token=%s level=%s error=%s", token, level, type(exc).__name__)
register_monitoring(RuleMonitoring)
# ---------------------------------------------------------------------------------- demo clock persistence
_writes: set[asyncio.Task[None]] = set()
async def _write_offset() -> None:
# Writes the offset current at write time, so out-of-order tasks still leave the latest value stored.
try:
async with get_clinical_session() as s:
await repo.save_clock_offset(s, clock.offset().total_seconds())
except (OSError, SQLAlchemyError) as exc:
log.warning("clock offset not saved error=%s", type(exc).__name__)
def _save_offset(_: timedelta) -> None:
"""Sync hook for `clock.register_persistence`: schedules the DB write on the running loop."""
try:
loop = asyncio.get_running_loop()
except RuntimeError:
log.warning("clock offset not saved: no running event loop")
return
task = loop.create_task(_write_offset())
_writes.add(task)
task.add_done_callback(_writes.discard)
async def flush_clock_writes() -> None:
"""Wait for pending offset writes (tests, shutdown)."""
if _writes:
await asyncio.gather(*list(_writes))
async def init_clock_persistence() -> datetime:
"""Restore the virtual clock offset from `clock_state` and save every later change there."""
async with get_clinical_session() as s:
stored = timedelta(seconds=await repo.load_clock_offset(s))
clock.register_persistence(lambda: stored, _save_offset)
log.info("clock offset restored seconds=%s", int(stored.total_seconds()))
return clock.now()
+93
View File
@@ -0,0 +1,93 @@
"""Patient snapshot for the rules engine, assembled from the clinical DB (ADR-007).
Maps storage names to `app.core.rules` names (`sleep_bucket` → `sleep_hours_bucket`, `alcohol_bucket` →
`alcohol`), derives age from `birth_year` and the therapy day from `therapy_start`. Time comes in as `as_of`
(from `app.core.clock`), never from the wall clock. Must not import `app.db.identity` (guardrail #5).
"""
from datetime import datetime, timedelta
from typing import Any, get_args
from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession
from app.core.rules import (
CssrsRisk,
DailyCheckin,
DoseRecord,
Module,
PatientSnapshot,
ScalePoint,
)
from app.db import repo
from app.db.clinical import Dose, Message, PatientProfile
# Risk screens are stored as scale results but reach the rules as `cssrs_risk`, not as scale points.
CSSRS_SCALES = frozenset({"cssrs_screen", "cssrs"})
# A C-SSRS result counts as the current risk for this long (same window as PHQ-9 item 9 and the dialog's
# C-SSRS cooldown): one old "high" must not restart the crisis protocol after the doctor has acted on it.
CSSRS_RECENT = timedelta(hours=24)
# Unknown sex/age fall back to the more cautious option: lower AUDIT-C cut-off, young-adult monitoring.
_UNKNOWN_SEX = "f"
_UNKNOWN_AGE = 0
_MODULES = frozenset(get_args(Module))
_CSSRS_RISKS = frozenset(get_args(CssrsRisk))
def _cssrs_risk(results: list[Any], as_of: datetime) -> CssrsRisk | None:
recent = [r for r in results if r.scale in CSSRS_SCALES and as_of - CSSRS_RECENT < r.ts <= as_of]
if not recent:
return None
severity = recent[-1].severity
return severity if severity in _CSSRS_RISKS else None
async def build_snapshot(
session: AsyncSession, token: str, as_of: datetime, *, profile: PatientProfile | None = None
) -> PatientSnapshot | None:
"""Snapshot of everything the rules may look at; None if the patient has no clinical profile."""
profile = profile or await repo.get_profile(session, token)
if profile is None:
return None
checkins = [
DailyCheckin(
day=c.day,
mood=c.mood,
med_taken=c.med_taken,
sleep_hours_bucket=c.sleep_bucket, # type: ignore[arg-type]
alcohol=c.alcohol_bucket, # type: ignore[arg-type]
side_effects=list(c.side_effects or []),
quit_intent=bool(c.quit_intent),
nightmares=c.nightmares,
)
for c in await repo.list_checkins(session, token)
]
results = [r for r in await repo.list_scale_results(session, token) if r.complete]
doses = (await session.scalars(select(Dose).where(Dose.token == token).order_by(Dose.planned_at))).all()
last_in = await session.scalar(
select(func.max(Message.ts)).where(Message.token == token, Message.direction == "in")
)
state = await repo.get_dialog_state(session, token)
facts = (state.data.get("facts") or {}) if state is not None else {}
return PatientSnapshot(
token=token,
as_of=as_of,
age=as_of.year - profile.birth_year if profile.birth_year else _UNKNOWN_AGE,
sex=profile.sex if profile.sex in ("m", "f") else _UNKNOWN_SEX, # type: ignore[arg-type]
modules={m for m in profile.modules or [] if m in _MODULES}, # type: ignore[misc]
therapy_day=(as_of.date() - profile.therapy_start).days if profile.therapy_start else 0,
checkins=checkins,
scale_results=[
ScalePoint(scale=r.scale, ts=r.ts, total=r.total, item9=r.item9, scheduled=r.scheduled)
for r in results
if r.scale not in CSSRS_SCALES
],
doses=[DoseRecord(planned_at=d.planned_at, status=d.status) for d in doses], # type: ignore[arg-type]
cssrs_risk=_cssrs_risk(results, as_of),
weapon_access=facts.get("weapon_access"),
last_patient_message_at=last_in,
open_alert_codes=await repo.open_alert_codes(session, token),
)
+3
View File
@@ -16,6 +16,9 @@ settings:
crisis_flags: [phq9_item9_positive, cssrs_high] crisis_flags: [phq9_item9_positive, cssrs_high]
# Шкала-скрининг суицидального риска (POSSIBLE и положительный п.9 PHQ-9 → эта шкала). # Шкала-скрининг суицидального риска (POSSIBLE и положительный п.9 PHQ-9 → эта шкала).
risk_screen_scale: cssrs_screen risk_screen_scale: cssrs_screen
# Брошенный шаг (чек-ин, шкала): если пациент молчит дольше N часов, следующее событие сначала возвращает
# диалог в idle (уже данные ответы сохранены), затем обрабатывается. 0 — выключено. На кризис не действует.
abandon_after_hours: 6
commands: commands:
# Текстовые команды отзыва согласия (сравниваются после приведения к нижнему регистру). # Текстовые команды отзыва согласия (сравниваются после приведения к нижнему регистру).
+98
View File
@@ -0,0 +1,98 @@
# Тексты приёма препарата (issue #37): напоминание, пропуск, побочные эффекты, вопрос о лекарстве, ответы
# на действия правил med_info (отсроченный эффект, алкоголь, побочные эффекты).
# СТАТУС: черновик, утверждает психиатр (#8).
#
# Правила текстов: без диагнозов и названий расстройств; без советов по дозам и схеме приёма (решает врач);
# не использовать слова, которые звучат как постановка на учёт или слежка (проверяется тестом). Ответы о
# препарате берутся только из выдержки инструкции ГРЛС (app/llm/med_qa.py); тексты ниже — обрамление и
# запасные варианты, если выдержки нет.
status: "черновик, утверждает психиатр (#8)"
reminder:
# {drug}, {dose} — из назначения врача (МНН и дозировка как в назначении, без изменений).
text: "Время принять {drug} {dose}. Отметьте, пожалуйста, как получилось."
text_no_drug: "Время принять препарат, назначенный врачом. Отметьте, пожалуйста, как получилось."
taken: "Принял(а)"
skip: "Пропущу"
side_effects: "Есть побочные эффекты"
taken_ack: "Отлично, отметил приём. Спасибо!"
# Кнопка от напоминания, которое уже неактуально (например, вчерашнего).
stale: "Это напоминание уже неактуально. Если хотите отметить самочувствие — нажмите «Как я сегодня»."
skip:
question: "Понял, отметил. Подскажите, что помешало? Это поможет вашему врачу."
reasons:
forgot: "Забыл(а)"
side_effects: "Из-за побочных эффектов"
quit: "Решил(а) бросить"
other: "Другое"
ack:
forgot: "Понял, бывает. Завтра напомню в обычное время."
quit: "Спасибо, что честно сказали — это важно, и решать это лучше вместе с врачом."
other: "Понял, отметил. Если захотите рассказать подробнее — просто напишите."
side_effects:
question: "Что беспокоит? Выберите вариант или «Другое», чтобы описать своими словами."
options:
nausea: "Тошнота"
insomnia: "Бессонница"
headache: "Головная боль"
dizziness: "Головокружение"
agitation: "Беспокойство"
other: "Другое"
# Короткий вопрос к инструкции препарата (med_qa, раздел «побочные эффекты») для выбранного эффекта.
questions:
nausea: "тошнота после приёма, когда пройдёт"
insomnia: "бессонница, нарушения сна"
headache: "головная боль"
dizziness: "головокружение"
agitation: "беспокойство, тревога"
describe: "Опишите, пожалуйста, коротко, что вы заметили."
ack: "Спасибо, отметил. Ваш врач увидит это в вашей карточке."
# Тот же эффект был и вчера: инструкцию второй раз не повторяем.
repeat: "Понимаю, это неприятно. Я отмечаю это каждый день, и ваш врач увидит, если не проходит."
ask_dose: "А препарат сегодня приняли?"
dose_yes: "Да, принял(а)"
dose_no: "Нет"
# Ежедневный чек-ин: если сегодня ещё нет отметки о приёме.
checkin:
question: "Сегодня приняли {drug}?"
question_no_drug: "Сегодня приняли назначенный препарат?"
"yes": "Да"
"no": "Нет"
later: "Ещё нет"
question:
prompt: "Напишите ваш вопрос о препарате. Я отвечаю только по официальной инструкции, а если ответа в ней нет — подскажу обратиться к врачу."
no_prescription: "Я не нашёл в карточке назначенного препарата. Пожалуйста, уточните вопрос у вашего врача."
# Ответы на действие правила med_info. Ответ = lead + выдержка из инструкции (med_qa по разделу section,
# вопрос question) + notice. Если выдержки нет — lead + fallback + notice.
med_info:
delayed_effect:
section: onset
question: "когда начнёт действовать препарат, нет эффекта, только хуже"
lead: "Понимаю, что сейчас тяжело."
fallback: "Эффект таких препаратов обычно наступает не сразу, а через 2–4 недели регулярного приёма. Пожалуйста, не прекращайте приём самостоятельно."
notice: "Я сообщил вашему врачу — он свяжется с вами, чтобы обсудить, как вы себя чувствуете."
alcohol_interaction:
section: alcohol
question: "можно ли алкоголь вместе с препаратом, сон"
lead: "Спасибо, что честно рассказали."
fallback: "В инструкции к препарату есть предупреждение о совместном приёме с алкоголем, а алкоголь сам по себе ухудшает сон."
notice: "Я передал это вашему врачу."
nausea:
section: side_effects
question: "тошнота после приёма, когда пройдёт"
lead: ""
fallback: "Тошнота часто встречается в первые дни приёма и обычно проходит сама."
notice: "Если станет хуже — напишите мне, я передам врачу."
side_effects:
section: side_effects
question: "побочные эффекты в начале приёма"
lead: ""
fallback: "Спасибо, что сообщили о самочувствии. Я отмечу это для вашего врача."
notice: "Если станет хуже — напишите мне, я передам врачу."
+62
View File
@@ -0,0 +1,62 @@
# Тексты ежедневного чек-ина и плановых шкал (issue #37, ADR-002, CLINICAL_MODULES §1–2).
# Ядро (настроение + приём) задаётся всем; вопрос про сон и кошмары — модулю veteran; алкоголь — см. settings.
# СТАТУС: черновик, утверждает психиатр (#8).
#
# Правила текстов: без диагнозов и названий расстройств; без советов по дозам; пациент не видит баллов шкал;
# не использовать слова, которые звучат как постановка на учёт или слежка (проверяется тестом).
status: "черновик, утверждает психиатр (#8)"
settings:
# Модули, которым задаётся вопрос про сон и кошмары (modules.yaml: daily_question = sleep_nightmares).
sleep_modules: [veteran]
# Ежедневный вопрос об алкоголе: модули, где он задаётся всегда (у ветеранов алкоголь — частый способ
# «заглушить» бессонницу и фактор риска, CLINICAL_MODULES §1); остальным — при положительном AUDIT-C
# (порог берётся из правила auditc_positive в config/rules.yaml, ADR-002).
alcohol_daily_modules: [veteran]
# Плановые шкалы, перед которыми показывается дружелюбное вступление (см. scale_intros).
intro_scales: [pcl5, phq9, dar5, auditc]
checkin:
start: "Добрый день! Короткая отметка о самочувствии — это займёт меньше минуты."
mood:
question: "Как ваше настроение сегодня по шкале от 0 до 10? 0 — очень плохо, 10 — отлично."
sleep:
question: "Сколько часов удалось поспать прошлой ночью?"
options:
"<4": "Меньше 4 ч"
"4-6": "4–6 ч"
"6-8": "6–8 ч"
">8": "Больше 8 ч"
nightmares:
question: "Снились ли кошмары или тяжёлые сны?"
"yes": "Да"
"no": "Нет"
alcohol:
question: "Выпивали ли вы алкоголь за последние сутки?"
clarify: "Вы упомянули алкоголь. Сколько примерно выпили за последние сутки?"
options:
none: "Не пил(а)"
"1-2": "1–2 порции"
more: "Больше"
text:
question: "Как вы? Можно пару слов — что было за день, что беспокоит."
skip: "Пропустить"
buttons_hint: "Выберите, пожалуйста, вариант кнопкой."
done: "Спасибо, записал. Если что-то изменится — пишите в любое время."
already_done: "Сегодня вы уже отмечались, спасибо! Если хотите что-то добавить — просто напишите."
visit:
# Нажата кнопка «Записаться к врачу раньше» (действие правила suggest_visit). Онлайн-записи пока нет —
# просьба передаётся врачу как факт в карточке.
requested: "Передал врачу просьбу о более раннем визите. С вами свяжутся, чтобы договориться о времени. Если станет хуже до визита — пишите мне или звоните 112."
# Дружелюбное вступление перед шкалой (параметр intro у start_scale). Пациент не видит баллов.
scale_intros:
scheduled:
pcl5: "Раз в две недели я прошу ответить на вопросы о том, как беспокоят тяжёлые воспоминания и их последствия. Это займёт 3–4 минуты. Так ваш врач увидит, что меняется со временем."
phq9: "Пора для короткого опроса о настроении и силах за последние две недели — 9 вопросов. Отвечайте так, как есть, правильных ответов нет."
dar5: "Несколько вопросов о раздражительности и вспышках гнева. После службы это бывает у многих, и врачу важно видеть, как это меняется."
auditc: "Три коротких вопроса об алкоголе. Отвечайте как есть: это нужно врачу, чтобы лечение было безопасным."
unscheduled:
phq9: "Последние дни настроение ниже обычного. Давайте я задам несколько вопросов — так ваш врач лучше поймёт, как вы сейчас."
+92
View File
@@ -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: |-
Помощник не выдаёт заключений и не отправляет ваши ответы ни в ГИБДД, ни в Росгвардию, ни на медкомиссии. Допуск к оружию и вождению решает отдельная медицинская комиссия по своим правилам, установленным законом.
Как именно это связано с вашим лечением — спросите лечащего врача прямо. Это нормальный вопрос, его задают многие, и честный ответ лучше, чем откладывать помощь из-за неизвестности.
+13 -3
View File
@@ -1,14 +1,24 @@
from fastapi.testclient import TestClient from fastapi.testclient import TestClient
from app.db import session
from app.db.bootstrap import init_db
from app.dialog.contracts import Event, get_engine from app.dialog.contracts import Event, get_engine
from app.main import app from app.main import app
from app.services.contracts import get_monitoring from app.services.contracts import get_monitoring
MEM = "sqlite+aiosqlite:///:memory:"
async def test_stubs_are_usable() -> None: async def test_stubs_are_usable() -> None:
replies = await get_engine().handle("tok", Event(type="text", text="привет")) await session.configure(identity_url=MEM, clinical_url=MEM)
assert replies and replies[0].text await init_db()
assert await get_monitoring().after_data_change("tok") == [] try:
replies = await get_engine().handle("tok", Event(type="text", text="привет"))
assert replies and replies[0].text
assert await get_monitoring().after_data_change("tok") == [] # unknown patient → no actions
assert await get_monitoring().is_in_crisis("tok") is False
finally:
await session.configure()
def test_api_fixture_routes() -> None: def test_api_fixture_routes() -> None:
+385
View File
@@ -0,0 +1,385 @@
"""Engine gaps found by the 13-day run (#56): no same-day C-SSRS repeat, scale intro registry, prefix triggers
with a button payload (medication buttons carry the dose), abandoned-step timeout."""
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.rules import Action
from app.core.scales import load_scale
from app.db import repo, session
from app.db.bootstrap import init_db
from app.db.clinical import Dose, ItemResponse, Prescription
from app.dialog import engine as engine_module
from app.dialog import fsm, scale_session
from app.dialog.contracts import Event, Reply
from app.dialog.engine import STEP_AT, 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, Step, get_trigger, is_registered, trigger, trigger_payload
from app.dialog.texts import load_dialog_texts
from app.llm.gateway import LLMGateway
from app.llm.mock import MockTasks
TOKEN = "g" * 32
OTHER = "h" * 32
MEM = "sqlite+aiosqlite:///:memory:"
NOW = datetime(2026, 10, 7, 9, 0)
TEXTS = load_dialog_texts()
MEDS = load_meds_texts()
DAILY = load_daily_texts()
SCREEN = TEXTS.settings.risk_screen_scale
class ScriptedMonitoring:
def __init__(self) -> None:
self.actions: list[Action] = []
async def after_data_change(self, token: str) -> list[Action]:
out, self.actions = self.actions, []
return out
async def is_in_crisis(self, token: str) -> bool:
return True
@pytest.fixture
async def mon(monkeypatch: pytest.MonkeyPatch) -> AsyncIterator[ScriptedMonitoring]:
monkeypatch.setattr(clock, "get_settings", lambda: Settings(_env_file=None, demo_clock=True))
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]
async def answer(e: Engine, scale_id: str, values: list[int]) -> list[Reply]:
out: list[Reply] = []
for item, value in zip(load_scale(scale_id).items, values, strict=False):
out += await press(e, f"scale:{scale_id}:{item.id}:{value}")
return out
def buttons(replies: list[Reply]) -> list[str]:
return [b.data for r in replies for row in r.buttons for b in row]
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(scale: str) -> list[Any]:
async with session.get_clinical_session() as s:
return list(await repo.list_scale_results(s, TOKEN, scale))
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())
# ----------------------------------------------------------------------------------------------- 1. C-SSRS repeat
async def test_item9_right_after_cssrs_does_not_repeat_screen(engine: Engine, mon: ScriptedMonitoring) -> None:
await system(engine, "scale_due", scale=SCREEN, scheduled=False)
assert (await get_state())[0] == f"scale.{SCREEN}"
await answer(engine, SCREEN, [0] * 6)
clock.advance(hours=1)
await system(engine, "scale_due", scale="phq9", scheduled=False)
# The rule phq9_item9 also asks for the screen — neither path may repeat it within the cooldown.
mon.actions = [Action(type="schedule_scale", code="phq9_item9", params={"scale": SCREEN, "scheduled": False})]
r = await answer(engine, "phq9", [1] * 8 + [1])
assert any(x.kind == "crisis" for x in r) # the fixed crisis script still goes out
state, data = await get_state()
assert state == "crisis" and data["crisis"]["open"] is True and not data.get("pending_scales")
assert len(await results(SCREEN)) == 1
async def test_item9_after_old_cssrs_asks_screen_again(engine: Engine) -> None:
await system(engine, "scale_due", scale=SCREEN, scheduled=False)
await answer(engine, SCREEN, [0] * 6)
clock.advance(hours=TEXTS.settings.cssrs_cooldown_hours + 1)
await system(engine, "scale_due", scale="phq9", scheduled=False)
r = await answer(engine, "phq9", [1] * 8 + [1])
assert any(x.kind == "crisis" for x in r)
assert (await get_state())[0] == f"scale.{SCREEN}" and TEXTS.scale.risk_screen_intro in r[-1].text
async def test_unfinished_cssrs_does_not_count(engine: Engine) -> None:
await system(engine, "scale_due", scale=SCREEN, scheduled=False)
await answer(engine, SCREEN, [0, 0]) # left halfway: no complete result
await system(engine, "scale_due", scale="phq9", scheduled=False) # queued behind the screen
async with session.get_clinical_session() as s:
assert not await scale_session.recent_risk_screen(_ctx(s), SCREEN)
def _ctx(s: Any, state: str = "idle", data: dict[str, Any] | None = None) -> Ctx:
return Ctx(
token=TOKEN,
state=state,
data=data or {},
now=clock.now(),
session=s,
gateway=LLMGateway(MockTasks()),
monitoring=ScriptedMonitoring(),
texts=TEXTS,
)
# ----------------------------------------------------------------------------------------------- 2. intros
async def test_registered_intro_text_and_callable(engine: Engine, monkeypatch: pytest.MonkeyPatch) -> None:
seen: list[tuple[str, bool, str]] = []
def intro(ctx: Ctx, scale_id: str, scheduled: bool, return_state: str) -> str | None:
seen.append((scale_id, scheduled, return_state))
return "Вступление к GAD-7"
monkeypatch.setattr(scale_session, "_intros", dict(scale_session._intros))
scale_session.register_scale_intro("gad7", intro)
with pytest.raises(ValueError):
scale_session.register_scale_intro("gad7", "второе")
r = await system(engine, "scale_due", scale="gad7", scheduled=True)
assert r[0].text.startswith("Вступление к GAD-7") and seen == [("gad7", True, "idle")]
scale_session.register_scale_intro("gad7", "Текстовое вступление", replace=True)
async with session.get_clinical_session() as s:
step = await scale_session.start_scale(_ctx(s), "gad7", return_state="idle")
assert step.data["scales"]["gad7"]["intro"] == "Текстовое вступление"
explicit = await scale_session.start_scale(_ctx(s), "gad7", return_state="idle", intro="Своё")
assert explicit.data["scales"]["gad7"]["intro"] == "Своё"
def test_daily_intros_use_registry_not_enter_hooks() -> None:
for scale_id in DAILY.settings.intro_scales:
assert not is_registered(f"scale.{scale_id}") # no per-scale enter hooks any more
assert scale_id in scale_session._intros
async def test_unscheduled_intro_from_registry(engine: Engine) -> None:
r = await system(engine, "scale_due", scale="phq9", scheduled=False)
assert DAILY.scale_intros.unscheduled["phq9"] in r[0].text
# ----------------------------------------------------------------------------------------------- 3. payload buttons
async def test_prefix_trigger_payload_and_precedence(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(fsm, "_triggers", dict(fsm._triggers))
monkeypatch.setattr(fsm, "_prefix_triggers", dict(fsm._prefix_triggers))
def handler(tag: str) -> Any:
async def fn(ctx: Ctx, event: Event) -> Step:
return Step(data={"tag": tag, "payload": trigger_payload(event)})
return fn
trigger("button:zz:a:*")(handler("short"))
trigger("button:zz:a:b:*")(handler("long"))
trigger("button:zz:a:exact")(handler("exact"))
with pytest.raises(ValueError):
trigger("button:zz:a:*")(handler("dup"))
async def run(key: str) -> dict[str, Any] | None:
fn = get_trigger(key)
if fn is None:
return None
return (await fn(None, Event(type="button", button=key.split(":", 1)[1]))).data # type: ignore[arg-type]
assert await run("button:zz:a:42") == {"tag": "short", "payload": "42"}
assert await run("button:zz:a:b:7:x") == {"tag": "long", "payload": "7:x"}
assert await run("button:zz:a:exact") == {"tag": "exact", "payload": None}
assert await run("button:zz:a:") is None and await run("button:zz:a") is None
async def test_reminder_buttons_carry_dose_and_no_state_key(engine: Engine) -> None:
async with session.get_clinical_session() as s:
await repo.set_dialog_state(
s, token=TOKEN, state="idle", data={"consent": True, "med_pending": {"dose_id": 1}}, now=clock.now()
)
r = await system(engine, "med_reminder")
[dose] = await doses()
assert buttons(r) == [f"med:taken:{dose.id}", f"med:skip:{dose.id}", f"med:side_effects:{dose.id}"]
assert "med_pending" not in (await get_state())[1]
skip = await press(engine, f"med:skip:{dose.id}")
assert f"med:skip_reason:side_effects:{dose.id}" in buttons(skip)
se = await press(engine, f"med:skip_reason:side_effects:{dose.id}")
await press(engine, f"med:se:nausea:{dose.id}")
[dose] = await doses()
assert dose.status == "skipped" and dose.side_effects == "nausea" and "med:se:nausea:" in buttons(se)[0]
async def test_two_reminders_each_button_answers_its_own_dose(engine: Engine) -> None:
first = await system(engine, "med_reminder")
second = await system(engine, "med_reminder")
await press(engine, buttons(first)[0]) # «Принял» of the FIRST reminder, pressed after the second arrived
d1, d2 = await doses()
assert (d1.status, d2.status) == ("taken", "pending")
await press(engine, buttons(second)[1])
assert [d.status for d in await doses()] == ["taken", "skipped"]
@pytest.mark.parametrize("data", ["med:taken", "med:skip", "med:se:nausea", "med:dose:yes", "med:skip_reason:quit"])
async def test_legacy_payloadless_buttons_are_stale(engine: Engine, data: str) -> None:
await system(engine, "med_reminder")
r = await press(engine, data)
assert r[0].text == MEDS.stale and await checkin() is None
assert [d.status for d in await doses()] == ["pending"]
async def test_foreign_unknown_and_old_dose_buttons_are_stale(engine: Engine) -> None:
async with session.get_clinical_session() as s:
rx = Prescription(token=OTHER, drug="сертралин", dose="50 мг", schedule={"times": ["08:00"]})
s.add(rx)
await s.flush()
foreign = Dose(token=OTHER, prescription_id=rx.id, planned_at=clock.now(), status="pending")
s.add(foreign)
await s.flush()
foreign_id = foreign.id
for data in (f"med:taken:{foreign_id}", "med:taken:9999", "med:taken:garbage", "med:taken:2026-10-06"):
assert (await press(engine, data))[0].text == MEDS.stale
old = await system(engine, "med_reminder")
clock.advance(days=1)
assert (await press(engine, buttons(old)[0]))[0].text == MEDS.stale
assert await checkin() is None
async def test_reminder_without_dose_row_uses_day_payload(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 system(engine, "med_reminder")
assert buttons(r)[0] == f"med:taken:{NOW.date().isoformat()}"
r = await press(engine, buttons(r)[0])
assert r[0].text == MEDS.taken_ack and (await checkin()).med_taken is True and await doses() == []
async def test_checkin_med_step_answers_todays_dose(engine: Engine) -> None:
await system(engine, "med_reminder")
await system(engine, "daily_checkin")
await press(engine, "daily:mood:5")
await press(engine, "daily:med:yes")
assert [d.status for d in await doses()] == ["taken"]
# ----------------------------------------------------------------------------------------------- 4. abandon timeout
async def test_abandoned_checkin_returns_to_idle(engine: Engine) -> None:
await system(engine, "daily_checkin")
await press(engine, "daily:mood:3")
assert (await get_state())[0].startswith("daily.")
clock.advance(hours=TEXTS.settings.abandon_after_hours + 1)
r = await system(engine, "med_reminder")
state, data = await get_state()
assert state == "idle" and "checkin" not in data
assert len(r) == 1 and r[0].text.startswith(MEDS.reminder.text.split("{")[0]) # no check-in question repeated
assert (await checkin()).mood == 3 # stored answer survives
async def test_short_pause_keeps_checkin(engine: Engine) -> None:
await system(engine, "daily_checkin")
clock.advance(hours=TEXTS.settings.abandon_after_hours - 1)
r = await press(engine, "daily:mood:6")
assert (await get_state())[0] != "idle" and r
async def test_abandoned_scale_lets_queued_scale_start(engine: Engine) -> None:
await system(engine, "scale_due", scale="phq9", scheduled=True)
await answer(engine, "phq9", [1, 2])
clock.advance(hours=4)
await system(engine, "scale_due", scale="pcl5", scheduled=True) # queued; system events don't extend the step
clock.advance(hours=3)
r = await text(engine, "добрый вечер")
state, data = await get_state()
assert state == "scale.pcl5" and "phq9" not in data["scales"] and not data.get("pending_scales")
assert DAILY.scale_intros.scheduled["pcl5"] in r[-1].text
assert await results("phq9") == [] # unfinished: no total, no severity
async with session.get_clinical_session() as s:
stored = (await s.scalars(select(ItemResponse).where(ItemResponse.scale == "phq9"))).all()
assert [x.score for x in stored] == [1, 2]
async def test_timeout_never_applies_in_crisis(engine: Engine) -> None:
await system(engine, "scale_due", scale="phq9", scheduled=False)
await answer(engine, "phq9", [0] * 8 + [2]) # item 9 → crisis + C-SSRS (none done before)
state, data = await get_state()
assert state == f"scale.{SCREEN}" and data["crisis"]["open"] is True
clock.advance(hours=TEXTS.settings.abandon_after_hours * 3)
await press(engine, f"scale:{SCREEN}:cssrs_1:0")
state, data = await get_state()
assert state == f"scale.{SCREEN}" and data["scales"][SCREEN]["idx"] == 1
async def test_timeout_never_skips_onboarding_scale(engine: Engine) -> None:
item = load_scale("pcl5").items[0]
old = (clock.now() - timedelta(hours=48)).isoformat()
session_data = {
"idx": 0,
"answers": {},
"return_state": "veteran.onboarding.phq9",
"scheduled": True,
"intro": None,
"intro_shown": True,
}
async with session.get_clinical_session() as s:
await repo.set_dialog_state(
s,
token=TOKEN,
state="scale.pcl5",
data={"consent": True, "scales": {"pcl5": session_data}, STEP_AT: old},
now=clock.now(),
)
await press(engine, f"scale:pcl5:{item.id}:1")
state, data = await get_state()
assert state == "scale.pcl5" and data["scales"]["pcl5"]["idx"] == 1
async def test_timeout_disabled_same_day(engine: Engine, monkeypatch: pytest.MonkeyPatch) -> None:
clock.set_now(NOW.replace(hour=1))
off = TEXTS.model_copy(update={"settings": TEXTS.settings.model_copy(update={"abandon_after_hours": 0})})
monkeypatch.setattr(engine_module, "load_dialog_texts", lambda: off)
await system(engine, "daily_checkin")
clock.advance(hours=20)
await press(engine, "daily:mood:4")
assert (await get_state())[0] != "idle" and (await checkin()).mood == 4
+377
View File
@@ -0,0 +1,377 @@
"""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)
+391
View File
@@ -0,0 +1,391 @@
"""Acceptance: 13 days of the veteran Andrey (docs/product/01_demo_persona.md) through the real dialog engine.
Andrey is set up post-onboarding directly in the DBs (no dependency on #36): modules veteran + depression,
therapy day 0, sertraline 50 mg at 08:00, consent. Each day: the dose reminder (as the scheduler would send it)
→ button; the daily check-in → buttons + the free text from the table.
Lane B's MonitoringService (#38) is replaced by a double that builds a `PatientSnapshot` from the clinical DB,
runs the REAL `app.core.rules.evaluate`, persists alert/crisis actions (so dedup works) and returns the dialog
actions — the same split as the frozen contract (ADR-007). Drug instruction excerpts are synthetic (FakeDrugInfo).
"""
from collections.abc import AsyncIterator
from datetime import date, datetime, timedelta
from typing import Any
import pytest
from sqlalchemy import func, select
from app.config import Settings
from app.core import clock
from app.core.rules import Action, DailyCheckin, DoseRecord, PatientSnapshot, ScalePoint, evaluate, patient_status
from app.core.scales import load_scale
from app.core.types import AlertLevel
from app.db import identity_repo, repo, session
from app.db.bootstrap import init_db
from app.db.clinical import Dose, Message, 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.integrations.base import DrugSection, integrations
from app.llm import med_qa as med_qa_module
from app.llm.gateway import LLMGateway
from app.llm.mock import MockTasks
from app.services import contracts as monitoring_contracts
TOKEN = "e" * 32
MEM = "sqlite+aiosqlite:///:memory:"
DAY0 = datetime(2026, 10, 5, 8, 0) # visit day; reminders at 08:00, check-ins at 20:00 (Europe/Moscow)
SRC = "ГРЛС: тестовая выдержка (синтетика)"
SERTRALINE = {
"side_effects": (
"Наиболее частые нежелательные реакции в начале терапии: тошнота, диарея, сухость во рту. "
"Тошнота обычно возникает в первые дни приёма и проходит в течение 1–2 недель. "
"Для уменьшения тошноты препарат принимают после еды."
),
"onset": (
"Терапевтический эффект развивается постепенно. "
"Улучшение обычно отмечается через 2–4 недели регулярного приёма. "
"Суточная доза может увеличиваться врачом до 200 мг."
),
"alcohol": (
"Совместный приём с алкоголем не рекомендуется. "
"Алкоголь может ухудшать сон и настроение и усиливать побочные эффекты."
),
}
# Persona table: day → (dose taken, mood, sleep, nightmares, alcohol, free text). "" = skip the free text.
TABLE: dict[int, tuple[bool, int, str, bool, str, str]] = {
1: (True, 4, "4-6", True, "none", ""),
2: (True, 4, "4-6", True, "none", "подташнивает с утра"),
3: (True, 4, "4-6", True, "none", "опять тошнит"),
4: (True, 3, "4-6", False, "none", "тошнота не проходит"),
5: (False, 3, "4-6", False, "none", "от таблеток только хуже, брошу"),
6: (True, 4, "6-8", False, "none", "врач позвонил, продолжаю"),
7: (True, 4, "6-8", False, "none", ""),
8: (True, 4, "<4", True, "1-2", "снится одно и то же"),
9: (True, 4, "<4", True, "more", ""),
10: (True, 4, "<4", True, "more", "без пива не могу уснуть"),
11: (True, 3, "4-6", False, "1-2", ""),
12: (True, 3, "4-6", False, "none", ""),
13: (True, 2, "<4", True, "none", "всё бессмысленно, устал от всего"),
}
# Expected dashboard colour at the end of each day (persona: 🟢🟢🟢🟡🟡🟢🟢🟢🟢🟡🟡🟡🔴).
COLORS = ["green"] * 3 + ["yellow"] * 2 + ["green"] * 4 + ["yellow"] * 3 + ["red"]
class FakeDrugInfo:
async def sections(self, inn: str) -> list[DrugSection]:
if inn != "сертралин":
return []
return [DrugSection(inn=inn, section=k, text=v, source=SRC) for k, v in SERTRALINE.items()]
class RuleMonitoring:
"""Test double of lane B's MonitoringService: real rules over a snapshot of the clinical DB."""
def __init__(self) -> None:
self.fired: list[str] = []
async def _snapshot(self, s: Any, token: str) -> PatientSnapshot:
now = clock.now()
profile = await repo.get_profile(s, token)
assert profile is not None and profile.therapy_start is not None and profile.birth_year is not None
checkins = [
DailyCheckin(
day=c.day,
mood=c.mood,
med_taken=c.med_taken,
sleep_hours_bucket=c.sleep_bucket, # type: ignore[arg-type]
alcohol=c.alcohol_bucket, # type: ignore[arg-type]
side_effects=list(c.side_effects or []),
quit_intent=bool(c.quit_intent),
nightmares=c.nightmares,
)
for c in await repo.list_checkins(s, token)
]
results = [r for r in await repo.list_scale_results(s, token) if r.complete]
doses = (await s.scalars(select(Dose).where(Dose.token == token, Dose.status != "pending"))).all()
cssrs = [r for r in results if r.scale == "cssrs_screen"]
last_in = await s.scalar(select(func.max(Message.ts)).where(Message.token == token, Message.direction == "in"))
return PatientSnapshot(
token=token,
as_of=now,
age=now.year - profile.birth_year,
sex=profile.sex, # type: ignore[arg-type]
modules=set(profile.modules), # type: ignore[arg-type]
therapy_day=(now.date() - profile.therapy_start).days,
checkins=checkins,
scale_results=[
ScalePoint(scale=r.scale, ts=r.ts, total=r.total, item9=r.item9, scheduled=r.scheduled)
for r in results
if r.scale != "cssrs_screen"
],
doses=[DoseRecord(planned_at=d.planned_at, status=d.status) for d in doses], # type: ignore[arg-type]
cssrs_risk=cssrs[-1].severity if cssrs else None, # type: ignore[arg-type]
weapon_access=False,
last_patient_message_at=last_in,
open_alert_codes=await repo.open_alert_codes(s, token),
)
async def after_data_change(self, token: str) -> list[Action]:
async with session.get_clinical_session() as s:
actions = evaluate(await self._snapshot(s, token))
for a in actions:
if a.creates_alert:
assert a.level is not None
await repo.add_alert(
s, token=token, level=a.level, code=a.code, reason=a.reason_ru, ts=clock.now(), params=a.params
)
self.fired += [a.code for a in actions]
return [a for a in actions if a.type != "alert"]
async def is_in_crisis(self, token: str) -> bool:
async with session.get_clinical_session() as s:
alerts = await repo.list_alerts(s, token=token, open_only=True)
return any(a.level == "acute" for a in alerts)
# ----------------------------------------------------------------------------------------------- fixtures
@pytest.fixture
async def andrey(monkeypatch: pytest.MonkeyPatch) -> AsyncIterator[RuleMonitoring]:
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(DAY0)
await session.configure(identity_url=MEM, clinical_url=MEM)
await init_db()
now = clock.now()
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-smirnova",
now=now,
birth_year=1992,
sex="m",
)
await identity_repo.record_consent(s, token=TOKEN, version="v1", now=now)
async with session.get_clinical_session() as s:
await repo.upsert_profile(
s,
token=TOKEN,
doctor_id="doc-smirnova",
modules=["veteran", "depression"],
therapy_start=now.date(),
next_visit=now.date() + timedelta(days=28),
birth_year=1992,
sex="m",
)
s.add(
Prescription(
token=TOKEN, drug="сертралин", dose="50 мг", schedule={"times": ["08:00"]}, start_date=now.date()
)
)
screening = now - timedelta(days=3) # day −3 screening (persona)
for scale, total in (("pcptsd5", 4), ("pcl5", 46), ("phq9", 14), ("auditc", 3)):
await repo.add_scale_result(
s,
token=TOKEN,
scale=scale,
total=total,
severity="-",
complete=True,
ts=screening,
item9=0 if scale == "phq9" else None,
)
await repo.set_dialog_state(s, token=TOKEN, state="idle", data={"consent": True, "adult": True}, now=now)
mon = RuleMonitoring()
monitoring_contracts.get_monitoring() # import lane B's module first if it exists, then override it
previous = monitoring_contracts._factory
monitoring_contracts.register_monitoring(lambda: mon)
yield mon
monitoring_contracts.register_monitoring(previous)
clock.reset()
await session.configure()
# ----------------------------------------------------------------------------------------------- helpers
class Andrey:
def __init__(self) -> None:
self.engine = Engine(gateway=LLMGateway(MockTasks())) # monitoring resolved via register_monitoring
async def text(self, t: str) -> list[Reply]:
return await self.engine.handle(TOKEN, Event(type="text", text=t))
async def press(self, data: str) -> list[Reply]:
return await self.engine.handle(TOKEN, Event(type="button", button=data))
async def system(self, kind: str, **params: Any) -> list[Reply]:
return await self.engine.handle(TOKEN, Event(type="system", system=kind, params=params)) # type: ignore[arg-type]
async def answer_scale(self, scale_id: str, values: list[int]) -> list[Reply]:
out: list[Reply] = []
for item, value in zip(load_scale(scale_id).items, values, strict=True):
out += await self.press(f"scale:{scale_id}:{item.id}:{value}")
return out
def buttons(replies: list[Reply]) -> list[str]:
return [b.data for r in replies for row in r.buttons for b in row]
def joined(replies: list[Reply]) -> str:
return "\n".join(r.text for r in replies)
async def 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 open_alerts() -> list[Any]:
async with session.get_clinical_session() as s:
return list(await repo.list_alerts(s, token=TOKEN, open_only=True))
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="doc-smirnova", ts=clock.now())
async def new_dose(day: date) -> int:
"""What the scheduler (#13) does before sending the reminder: a planned dose row."""
async with session.get_clinical_session() as s:
rx = await s.scalar(select(Prescription).where(Prescription.token == TOKEN))
assert rx is not None
row = Dose(token=TOKEN, prescription_id=rx.id, planned_at=clock.now(), status="pending")
s.add(row)
await s.flush()
return row.id
# ----------------------------------------------------------------------------------------------- the 13 days
async def test_andrey_13_days(andrey: RuleMonitoring) -> None:
a = Andrey()
meds = load_meds_texts()
daily = load_daily_texts()
log: dict[int, list[Reply]] = {}
colors: list[str] = []
for day in range(1, 14):
taken, mood, sleep, nightmares, alcohol, free_text = TABLE[day]
clock.set_now(DAY0 + timedelta(days=day))
if day == 6:
await ack_all() # the doctor called Andrey after the day-5 alerts and acknowledged them
r: list[Reply] = []
# --- 08:00 dose reminder
dose_id = await new_dose(clock.today())
reminder = await a.system("med_reminder", dose_id=dose_id, drug="сертралин", dose="50 мг")
assert "сертралин 50 мг" in reminder[0].text
assert buttons(reminder) == [f"med:taken:{dose_id}", f"med:skip:{dose_id}", f"med:side_effects:{dose_id}"]
if taken:
r += await a.press(f"med:taken:{dose_id}")
else:
skip = await a.press(f"med:skip:{dose_id}")
assert f"med:skip_reason:quit:{dose_id}" in buttons(skip)
r += skip + await a.press(f"med:skip_reason:quit:{dose_id}")
# --- day 7: scheduled scales (PCL-5 = 44, PHQ-9 = 13, item 9 = 0) with friendly intros
if day == 7:
first = await a.system("scale_due", scale="pcl5", scheduled=True)
assert daily.scale_intros.scheduled["pcl5"] in first[0].text
await a.system("scale_due", scale="phq9", scheduled=True) # queued behind PCL-5
after_pcl = await a.answer_scale("pcl5", [2] * 16 + [3] * 4)
assert daily.scale_intros.scheduled["phq9"] in after_pcl[-1].text
await a.answer_scale("phq9", [2] * 5 + [1] * 3 + [0])
assert (await state())[0] == "idle"
# --- 20:00 daily check-in
clock.set_now(DAY0 + timedelta(days=day, hours=12))
start = await a.system("daily_checkin")
assert start[0].text == daily.checkin.start and buttons(start)[0] == "daily:mood:0"
r += await a.press(f"daily:mood:{mood}") # dose already answered by the reminder → no med question
assert "daily:sleep:<4" in buttons(r[-1:])
r += await a.press(f"daily:sleep:{sleep}")
r += await a.press(f"daily:nightmares:{'yes' if nightmares else 'no'}")
r += await a.press(f"daily:alcohol:{alcohol}")
assert buttons(r[-1:]) == ["daily:text:skip"]
r += await a.text(free_text) if free_text else await a.press("daily:text:skip")
log[day] = r
if day == 10: # «записаться раньше» offered by rule alcohol_3d
visit = await a.press("menu:visit")
assert visit[0].text == daily.visit.requested and (await state())[1]["visit_requests"]
if day == 13:
break # the crisis path is checked below
assert (await state())[0] == "idle"
colors.append(patient_status(AlertLevel[x.level.upper()] for x in await open_alerts()).color)
# Day 2: first nausea → instruction excerpt (med_qa, section side_effects), not generated advice.
assert "По инструкции к препарату" in joined(log[2]) and "проходит в течение 1–2 недель" in joined(log[2])
# Day 3: same effect again → short support, the instruction is not repeated.
assert meds.side_effects.repeat in joined(log[3]) and "По инструкции" not in joined(log[3])
# Day 5: «решил бросить» → rule quit_intent → delayed onset + do not stop on your own + doctor notified.
day5 = joined(log[5])
assert "2–4 недели" in day5 and "не прекращайте" in day5 and meds.med_info["delayed_effect"].notice in day5
assert "200 мг" not in day5 # dosing sentences of the instruction are never quoted
# Day 10: sleep < 4 h ×3 + alcohol ×3 → interaction info + «записаться раньше».
day10 = joined(log[10])
assert "Совместный приём с алкоголем не рекомендуется" in day10
assert buttons(log[10])[-1] == "menu:visit"
# Day 11: alerts still open → the rule stays silent (no repeated interaction text).
assert "Совместный приём с алкоголем" not in joined(log[11]) and "menu:visit" not in buttons(log[11])
assert colors == COLORS[:12]
# Day 13: «всё бессмысленно» → POSSIBLE → C-SSRS screen (buttons only) took over the free text.
day13 = log[13]
assert day13[-1].kind == "question" and "scale:cssrs_screen:cssrs_1:0" in buttons(day13[-1:])
r = await a.answer_scale("cssrs_screen", [0] * 6)
# Back in the check-in: the message counts as the answer → check-in closed → mood ≤ 3 three days →
# unscheduled PHQ-9 with its own intro.
assert daily.checkin.done in joined(r) and daily.scale_intros.unscheduled["phq9"] in r[-1].text
st, data = await state()
assert st == "scale.phq9" and data["scales"]["phq9"]["scheduled"] is False
r = await a.answer_scale("phq9", [2] * 8 + [2]) # total 18, item 9 = 2
assert any(x.kind == "crisis" for x in r) and any(x.kind == "safety_plan" for x in r)
st, data = await state()
# C-SSRS was completed minutes ago: item 9 opens the crisis but does not ask the same screen again (#56).
assert data["crisis"]["open"] is True and st == "crisis" and not data.get("pending_scales")
async with session.get_clinical_session() as s:
assert len(await repo.list_scale_results(s, TOKEN, "cssrs_screen")) == 1
acute = [x for x in await open_alerts() if x.level == "acute"]
assert [x.code for x in acute] == ["phq9_item9"]
assert patient_status(AlertLevel[x.level.upper()] for x in await open_alerts()).color == COLORS[12]
# Stored check-ins match the persona table.
async with session.get_clinical_session() as s:
rows = {c.day: c for c in await repo.list_checkins(s, TOKEN)}
doses = (await s.scalars(select(Dose).where(Dose.token == TOKEN).order_by(Dose.id))).all()
phq = list(await repo.list_scale_results(s, TOKEN, "phq9"))
for day, (taken, mood, sleep, nightmares, alcohol, _text) in TABLE.items():
c = rows[(DAY0 + timedelta(days=day)).date()]
assert (c.med_taken, c.mood, c.sleep_bucket, c.nightmares, c.alcohol_bucket) == (
taken,
mood,
sleep,
nightmares,
alcohol,
), day
for day in (2, 3, 4):
assert "nausea" in rows[(DAY0 + timedelta(days=day)).date()].side_effects
assert rows[(DAY0 + timedelta(days=5)).date()].quit_intent is True
assert rows[(DAY0 + timedelta(days=13)).date()].free_text_scrubbed is None # taken over by the C-SSRS
assert [d.status for d in doses] == ["skipped" if d == 5 else "taken" for d in range(1, 14)]
assert [(p.total, p.item9, p.scheduled) for p in phq] == [(14, 0, True), (13, 0, True), (18, 2, False)]
for code in ("side_effect_3d", "quit_intent", "sleep_lt4_3n", "alcohol_3d", "mood_low_3d", "phq9_item9"):
assert code in andrey.fired, code
+368
View File
@@ -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}»"
+279
View File
@@ -0,0 +1,279 @@
"""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
"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", "high")]
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)