From d06720fdab2b14eb11e9c5249d705553e9f3d110 Mon Sep 17 00:00:00 2001 From: Lacky1234union Date: Sat, 3 Oct 2026 19:00:47 +0300 Subject: [PATCH] fix(bot): close dialog engine gaps (C-SSRS repeat, scale intros, button payloads, step timeout) - 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::*" with trigger_payload(event); medication buttons carry the dose (med:taken:, ...), 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 --- app/dialog/engine.py | 53 +++- app/dialog/flows/meds.py | 180 +++++++++----- app/dialog/flows/veteran_daily.py | 47 ++-- app/dialog/fsm.py | 61 ++++- app/dialog/scale_session.py | 62 ++++- app/dialog/texts.py | 1 + config/texts/dialog.yaml | 3 + tests/test_dialog_engine_gaps.py | 385 ++++++++++++++++++++++++++++++ tests/test_dialog_meds.py | 65 ++--- tests/test_dialog_veteran_days.py | 15 +- 10 files changed, 733 insertions(+), 139 deletions(-) create mode 100644 tests/test_dialog_engine_gaps.py diff --git a/app/dialog/engine.py b/app/dialog/engine.py index e959d6c..c76543f 100644 --- a/app/dialog/engine.py +++ b/app/dialog/engine.py @@ -2,6 +2,10 @@ 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 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 @@ -25,7 +29,7 @@ Logs carry only the token, states and event codes (guardrail #8). import asyncio import logging from dataclasses import dataclass, field -from datetime import datetime +from datetime import datetime, timedelta from typing import Any 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.flows import load_flows 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.llm.gateway import LLMGateway, get_gateway from app.services import questionnaire as intake_service @@ -58,6 +71,8 @@ MAX_ENTER_CHAIN = 8 _patient_locks = WeakValueDictionary() # DB unreachable: answer safely instead of failing the channel (connectivity errors only — bugs must surface). _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 @@ -109,6 +124,8 @@ class Engine: state_before = ctx.state run = _Run() + if self._abandoned(ctx): + await self._apply(ctx, self._abandon(ctx), run) await self._apply(ctx, await self._route(ctx, event), run) if run.notify: @@ -126,6 +143,9 @@ class Engine: await repo.add_message( 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) task_id = event.params.get("questionnaire_task") if event.type == "system" and task_id and run.replies: @@ -145,6 +165,24 @@ class Engine: ) 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: consented = _consented(ctx) if event.type == "text": @@ -307,10 +345,19 @@ class Engine: if not pending or ctx.state not in RESTING_STATES: return Step() 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": pending_screen = [p for p in pending if p["scale"] == screen] if not pending_screen: - return Step() + return Step(data=dropped) item = pending_screen[0] else: item = pending[0] diff --git a/app/dialog/flows/meds.py b/app/dialog/flows/meds.py index f848ed0..d079160 100644 --- a/app/dialog/flows/meds.py +++ b/app/dialog/flows/meds.py @@ -14,8 +14,12 @@ Everything the patient answers is written to the clinical DB (`doses`, `daily_ch 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 fixed data ("med:taken", ...) so they are plain idle triggers; the dose they refer to is -kept in `dialog_state.data["med_pending"]` ({dose_id, drug, dose, day}) — a button from an earlier day is stale. +Reminder buttons carry the dose they refer to (#56): "med:taken:", "med:skip:", +"med:side_effects:" and the follow-up buttons likewise ("med:se::", "med:skip_reason::", +"med:dose:yes:"), matched by prefix triggers. `` 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 @@ -36,7 +40,7 @@ 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, get_trigger, handler_for, state, trigger +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 @@ -158,36 +162,55 @@ 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 pending_dose(ctx: Ctx) -> dict[str, Any] | None: - """Today's reminder the patient is answering, or None (no reminder / reminder from an earlier day).""" - pending = ctx.data.get("med_pending") - if isinstance(pending, dict) and pending.get("day") == ctx.now.date().isoformat(): - return pending - return None +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 _dose_row(ctx: Ctx, pending: dict[str, Any] | None) -> Dose | None: - if not pending or pending.get("dose_id") is None: - return None - row = await ctx.session.get(Dose, pending["dose_id"]) - return row if row is not None and row.token == ctx.token else None +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 record_dose(ctx: Ctx, taken: bool, *, status: str | None = None) -> None: - """Dose answer → `doses` row of today's reminder (if any) + `daily_checkins.med_taken` (rules read both).""" - row = await _dose_row(ctx, pending_dose(ctx)) +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]) -> None: +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, pending_dose(ctx)) + 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) @@ -228,12 +251,12 @@ async def run_trigger_inline(ctx: Ctx, key: str, event: Event) -> Step | None: # ------------------------------------------------------------------------------------------- reminder -def _reminder_buttons() -> list[list[Button]]: +def _reminder_buttons(ref: str) -> list[list[Button]]: t = load_meds_texts().reminder return [ - [Button(label=t.taken, data="med:taken")], - [Button(label=t.skip, data="med:skip")], - [Button(label=t.side_effects, data="med:side_effects")], + [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}")], ] @@ -253,43 +276,68 @@ async def _med_reminder(ctx: Ctx, event: Event) -> Step: ctx.session.add(row) await ctx.session.flush() dose_id = row.id - pending = {"dose_id": dose_id, "drug": drug, "dose": dose, "day": ctx.now.date().isoformat()} text = t.text.format(drug=drug, dose=dose).replace(" ", " ") if drug else t.text_no_drug - return Step(replies=[Reply(text=text, buttons=_reminder_buttons(), kind="question")], data={"med_pending": pending}) + 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(ctx: Ctx) -> Step | None: - return None if pending_dose(ctx) is not None else Step(replies=[Reply(text=load_meds_texts().stale)]) +def _stale_reply() -> Step: + return Step(replies=[Reply(text=load_meds_texts().stale)]) -@trigger("button:med:taken") -async def _taken(ctx: Ctx, event: Event) -> Step: - if (stale := _stale(ctx)) is not None: - return stale - await record_dose(ctx, True) +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::*` (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) -@trigger("button:med:skip") -async def _skip(ctx: Ctx, event: Event) -> Step: - if (stale := _stale(ctx)) is not None: - return stale - await record_dose(ctx, False, status="skipped") +@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}")] for r in SKIP_REASONS] + 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() -> Reply: +def _side_effect_question(ref: str) -> Reply: t = load_meds_texts().side_effects - buttons = [[Button(label=t.options[c], data=f"med:se:{c}")] for c in SIDE_EFFECT_CODES] + 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) -> Step: + async def handler(ctx: Ctx, event: Event, ref: str) -> Step: if reason == "side_effects": - return Step(replies=[_side_effect_question()]) + 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). @@ -301,53 +349,62 @@ def _skip_reason(reason: str) -> Any: for _reason in SKIP_REASONS: - trigger(f"button:med:skip_reason:{_reason}")(_skip_reason(_reason)) + med_trigger(f"skip_reason:{_reason}")(_skip_reason(_reason)) -@trigger("button:med:side_effects") -async def _side_effects(ctx: Ctx, event: Event) -> Step: - return Step(replies=[_side_effect_question()]) +@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]) -> Step: - await record_side_effects(ctx, codes) +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="med:dose:yes")], [Button(label=t.dose_no, data="med:dose:no")]] + 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", notify_monitoring=True) + 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) -> Step: + 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 + 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]) + return await _after_side_effects(ctx, [code], ref) return handler for _code in SIDE_EFFECT_CODES: - trigger(f"button:med:se:{_code}")(_side_effect_code(_code)) + 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) + 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"] - return await _after_side_effects(ctx, codes) + 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) -> Step: - await record_dose(ctx, taken, status=None if taken else "missed") + 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, @@ -356,8 +413,11 @@ def _dose_answer(taken: bool) -> Any: return handler -trigger("button:med:dose:yes")(_dose_answer(True)) -trigger("button:med:dose:no")(_dose_answer(False)) +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 diff --git a/app/dialog/flows/veteran_daily.py b/app/dialog/flows/veteran_daily.py index 8cd8873..1ea496c 100644 --- a/app/dialog/flows/veteran_daily.py +++ b/app/dialog/flows/veteran_daily.py @@ -24,14 +24,14 @@ Data algorithm (medical data, jury requirement): 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 system event of a later day. + * 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) — the engine starts `scale_due` -scales without an intro, so an exact `scale.` enter hook adds one and delegates to the generic scale hook. +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 dataclasses import replace from datetime import datetime from functools import lru_cache from pathlib import Path @@ -59,9 +59,8 @@ from app.dialog.flows.meds import ( save_checkin, side_effect_reply, ) -from app.dialog.fsm import Ctx, Step, enter_hook_for, on_enter, state, trigger -from app.dialog.scale_session import PREFIX as SCALE_PREFIX -from app.dialog.scale_session import active_scale, scale_state +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") @@ -346,6 +345,9 @@ async def _free_text(ctx: Ctx, ck: dict[str, Any]) -> Step: 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 "" @@ -451,30 +453,13 @@ async def _idle_text(ctx: Ctx, event: Event) -> Step: # ------------------------------------------------------------------------------------------- scale intros -async def _scale_intro(ctx: Ctx) -> Step: - """Add the friendly intro of a scheduled (or rule-triggered) scale, then run the generic scale enter hook. - - Onboarding sessions (#36) and sessions started with their own intro are left untouched. - """ - scale_id = active_scale(ctx.state) - sessions = dict(ctx.data.get("scales") or {}) - session = sessions.get(scale_id) if scale_id else None - if ( - scale_id is not None - and session is not None - and session.get("idx") == 0 - and not session.get("intro_shown") - and not session.get("intro") - and ".onboarding" not in str(session.get("return_state", "")) - ): - intros = load_daily_texts().scale_intros - intro = (intros.scheduled if session.get("scheduled", True) else intros.unscheduled).get(scale_id) - if intro: - sessions[scale_id] = {**session, "intro": intro} - ctx = replace(ctx, data={**ctx.data, "scales": sessions}) - generic = enter_hook_for(SCALE_PREFIX) - return await generic(ctx) if generic is not None else Step() +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: - on_enter(scale_state(_scale_id))(_scale_intro) + register_scale_intro(_scale_id, _scale_intro) diff --git a/app/dialog/fsm.py b/app/dialog/fsm.py index e1b3860..e576b1e 100644 --- a/app/dialog/fsm.py +++ b/app/dialog/fsm.py @@ -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 with menu buttons, system events and free text; the core registers defaults, a flow overrides with `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 @@ -103,6 +111,11 @@ EnterHook = Callable[[Ctx], Awaitable[Step]] _handlers: dict[str, Handler] = {} _enter_hooks: dict[str, EnterHook] = {} _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]: @@ -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. Keys: "button:", "system:", "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: - 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)") - _triggers[key] = fn + table[name] = fn return fn 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: candidate = name while candidate: @@ -167,7 +207,20 @@ def enter_hook_for(name: str) -> EnterHook | 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: diff --git a/app/dialog/scale_session.py b/app/dialog/scale_session.py index 366eded..48c3657 100644 --- a/app/dialog/scale_session.py +++ b/app/dialog/scale_session.py @@ -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. 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 - once: partial result saved, fixed crisis script, C-SSRS queued (unless this *is* the C-SSRS). A positive - item 9 is acted upon even in a half-finished questionnaire. + once: partial result saved, fixed crisis script, C-SSRS queued (unless this *is* the C-SSRS, or a C-SSRS was + 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 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. + +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.` itself. """ +from collections.abc import Callable +from dataclasses import replace +from datetime import timedelta from typing import Any 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.dialog.contracts import Button, Event, Reply 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" +# (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: 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.`. `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 + if intro is None: + intro = scale_intro(ctx, scale_id, scheduled=scheduled, return_state=return_state) sessions = dict(ctx.data.get("scales") or {}) sessions[scale_id] = { "idx": 0, @@ -98,6 +127,28 @@ def _session(ctx: Ctx) -> tuple[ScaleDef, dict[str, Any]] | None: 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) async def _enter(ctx: Ctx) -> Step: found = _session(ctx) @@ -231,6 +282,7 @@ async def _finish( else: crisis = open_crisis(ctx, categories=[], source=f"scale:{scale.id}") 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) return step diff --git a/app/dialog/texts.py b/app/dialog/texts.py index 6723cb8..c209dac 100644 --- a/app/dialog/texts.py +++ b/app/dialog/texts.py @@ -19,6 +19,7 @@ class DialogSettings(_Strict): max_safety_events: int = 50 crisis_flags: list[str] = Field(default_factory=list) risk_screen_scale: str = "cssrs_screen" + abandon_after_hours: float = 6 class Commands(_Strict): diff --git a/config/texts/dialog.yaml b/config/texts/dialog.yaml index c2b58cb..9361d70 100644 --- a/config/texts/dialog.yaml +++ b/config/texts/dialog.yaml @@ -16,6 +16,9 @@ settings: crisis_flags: [phq9_item9_positive, cssrs_high] # Шкала-скрининг суицидального риска (POSSIBLE и положительный п.9 PHQ-9 → эта шкала). risk_screen_scale: cssrs_screen + # Брошенный шаг (чек-ин, шкала): если пациент молчит дольше N часов, следующее событие сначала возвращает + # диалог в idle (уже данные ответы сохранены), затем обрабатывается. 0 — выключено. На кризис не действует. + abandon_after_hours: 6 commands: # Текстовые команды отзыва согласия (сравниваются после приведения к нижнему регистру). diff --git a/tests/test_dialog_engine_gaps.py b/tests/test_dialog_engine_gaps.py new file mode 100644 index 0000000..474cf5c --- /dev/null +++ b/tests/test_dialog_engine_gaps.py @@ -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 diff --git a/tests/test_dialog_meds.py b/tests/test_dialog_meds.py index cc48de2..803ac9d 100644 --- a/tests/test_dialog_meds.py +++ b/tests/test_dialog_meds.py @@ -101,6 +101,17 @@ 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) @@ -125,10 +136,10 @@ async def doses() -> list[Dose]: 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 мг") - assert buttons(r) == ["med:taken", "med:skip", "med:side_effects"] [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 press(engine, "med:taken") + 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 @@ -136,18 +147,17 @@ async def test_reminder_without_dose_id_creates_dose_and_taken(engine: Engine, m async def test_stale_reminder_button(engine: Engine) -> None: - await system(engine, "med_reminder") + reminder = await system(engine, "med_reminder") clock.advance(days=1) - r = await press(engine, "med:taken") + 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: - await system(engine, "med_reminder") - r = await press(engine, "med:skip") - assert buttons(r) == [f"med:skip_reason:{x}" for x in ("forgot", "side_effects", "quit", "other")] - r = await press(engine, f"med:skip_reason:{reason}") + 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 @@ -155,44 +165,39 @@ async def test_skip_simple_reasons(engine: Engine, reason: str) -> None: async def test_skip_quit_sets_intent_and_explains(engine: Engine, mon: ScriptedMonitoring) -> None: - await system(engine, "med_reminder") - await press(engine, "med:skip") + 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 press(engine, "med:skip_reason:quit") + 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: - await system(engine, "med_reminder") - await press(engine, "med:skip") - r = await press(engine, "med:skip_reason:side_effects") - assert "med:se:nausea" in buttons(r) and "med:se:other" in buttons(r) + 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: - await system(engine, "med_reminder") - r = await press(engine, "med:side_effects") - assert buttons(r) == [f"med:se:{c}" for c in ("nausea", "insomnia", "headache", "dizziness", "agitation", "other")] - r = await press(engine, "med:se:nausea") + 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 buttons(r[1:]) == ["med:dose:yes", "med:dose:no"] + 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 press(engine, "med:dose:yes") + 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: - await system(engine, "med_reminder") - await press(engine, "med:side_effects") - r = await press(engine, "med:se:other") + 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 press(engine, "med:side_effects") - await press(engine, "med:se:other") + await tap(engine, await tap(engine, reminder, "med:side_effects"), "med:se:other") await text(engine, "какое-то странное ощущение") assert (await checkin()).side_effects == ["headache", "other"] @@ -202,8 +207,8 @@ async def test_same_side_effect_as_yesterday_gets_support(engine: Engine) -> Non await repo.upsert_checkin( s, token=TOKEN, day=clock.today() - timedelta(days=1), now=clock.now(), side_effects=["nausea"] ) - await system(engine, "med_reminder") - r = await press(engine, "med:se: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 @@ -276,8 +281,8 @@ async def test_checkin_asks_dose_when_not_answered_and_typed_mood(engine: Engine async def test_reminder_during_checkin_keeps_checkin(engine: Engine) -> None: await system(engine, "daily_checkin") r = await system(engine, "med_reminder") - assert buttons(r[:1]) == ["med:taken", "med:skip", "med:side_effects"] and r[-1].text == DAILY.checkin.mood.question - r = await press(engine, "med:taken") + 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 diff --git a/tests/test_dialog_veteran_days.py b/tests/test_dialog_veteran_days.py index bd73b02..f3ba2fa 100644 --- a/tests/test_dialog_veteran_days.py +++ b/tests/test_dialog_veteran_days.py @@ -292,13 +292,13 @@ async def test_andrey_13_days(andrey: RuleMonitoring) -> None: 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) == ["med:taken", "med:skip", "med:side_effects"] + 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("med:taken") + r += await a.press(f"med:taken:{dose_id}") else: - skip = await a.press("med:skip") - assert "med:skip_reason:quit" in buttons(skip) - r += skip + await a.press("med:skip_reason:quit") + 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: @@ -359,7 +359,10 @@ async def test_andrey_13_days(andrey: RuleMonitoring) -> None: 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() - assert data["crisis"]["open"] is True and st in ("crisis", "scale.cssrs_screen") + # 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]