fix(bot): пробелы движка — повтор C-SSRS, вступления шкал, данные в кнопках, тайм-аут шага #57

Merged
Eleboba merged 1 commits from fix/56-engine-gaps into dev 2026-10-03 19:03:11 +03:00
10 changed files with 733 additions and 139 deletions
+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]
+120 -60
View File
@@ -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 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. (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 Reminder buttons carry the dose they refer to (#56): "med:taken:<ref>", "med:skip:<ref>",
kept in `dialog_state.data["med_pending"]` ({dose_id, drug, dose, day}) — a button from an earlier day is stale. "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 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.actions import register_med_info_handler
from app.dialog.contracts import Button, Event, Reply from app.dialog.contracts import Button, Event, Reply
from app.dialog.flows.core import menu_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.integrations.base import integrations
from app.llm.med_qa import fallback_text, med_qa 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) 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: def dose_ref(dose_id: int | None, day: date) -> str:
"""Today's reminder the patient is answering, or None (no reminder / reminder from an earlier day).""" """Button payload naming a dose: its row id, or the reminder day when there is no row."""
pending = ctx.data.get("med_pending") return str(dose_id) if dose_id is not None else day.isoformat()
if isinstance(pending, dict) and pending.get("day") == ctx.now.date().isoformat():
return pending
return None
async def _dose_row(ctx: Ctx, pending: dict[str, Any] | None) -> Dose | None: async def todays_dose(ctx: Ctx) -> Dose | None:
if not pending or pending.get("dose_id") is None: """Latest dose row planned for today (local day of `ctx.now`) — target of answers given without a button."""
return None rows = await ctx.session.scalars(select(Dose).where(Dose.token == ctx.token).order_by(Dose.id.desc()).limit(20))
row = await ctx.session.get(Dose, pending["dose_id"]) today = ctx.now.date()
return row if row is not None and row.token == ctx.token else None 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: async def resolve_ref(ctx: Ctx, ref: str | None) -> tuple[bool, Dose | None]:
"""Dose answer → `doses` row of today's reminder (if any) + `daily_checkins.med_taken` (rules read both).""" """(fresh, dose row) for a button payload. Not fresh: missing/foreign/unknown ref or a dose of another day."""
row = await _dose_row(ctx, pending_dose(ctx)) 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: if row is not None:
row.status = status or ("taken" if taken else "skipped") row.status = status or ("taken" if taken else "skipped")
row.answered_at = ctx.now row.answered_at = ctx.now
await save_checkin(ctx, med_taken=taken) 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).""" """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()) today = await checkin_of(ctx, ctx.now.date())
merged = list(dict.fromkeys([*(today.side_effects if today and today.side_effects else []), *codes])) merged = list(dict.fromkeys([*(today.side_effects if today and today.side_effects else []), *codes]))
await save_checkin(ctx, side_effects=merged) 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: if row is not None:
row.side_effects = ",".join(c for c in dict.fromkeys([*(row.side_effects or "").split(","), *codes]) if c) 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 # ------------------------------------------------------------------------------------------- reminder
def _reminder_buttons() -> list[list[Button]]: def _reminder_buttons(ref: str) -> list[list[Button]]:
t = load_meds_texts().reminder t = load_meds_texts().reminder
return [ return [
[Button(label=t.taken, data="med:taken")], [Button(label=t.taken, data=f"med:taken:{ref}")],
[Button(label=t.skip, data="med:skip")], [Button(label=t.skip, data=f"med:skip:{ref}")],
[Button(label=t.side_effects, data="med:side_effects")], [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) ctx.session.add(row)
await ctx.session.flush() await ctx.session.flush()
dose_id = row.id 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 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: def _stale_reply() -> Step:
return None if pending_dose(ctx) is not None else Step(replies=[Reply(text=load_meds_texts().stale)]) return Step(replies=[Reply(text=load_meds_texts().stale)])
@trigger("button:med:taken") async def _legacy(ctx: Ctx, event: Event) -> Step:
async def _taken(ctx: Ctx, event: Event) -> Step: """Payload-less medication button of an earlier release: the dose is unknown → treat as stale."""
if (stale := _stale(ctx)) is not None: return _stale_reply()
return stale
await record_dose(ctx, True)
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) return Step(replies=[Reply(text=load_meds_texts().taken_ack)], notify_monitoring=True)
@trigger("button:med:skip") @med_trigger("skip")
async def _skip(ctx: Ctx, event: Event) -> Step: async def _skip(ctx: Ctx, event: Event, ref: str) -> Step:
if (stale := _stale(ctx)) is not None: await record_dose(ctx, False, status="skipped", ref=ref)
return stale
await record_dose(ctx, False, status="skipped")
t = load_meds_texts().skip 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) 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 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") return Reply(text=t.question, buttons=buttons, kind="question")
def _skip_reason(reason: str) -> Any: 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": if reason == "side_effects":
return Step(replies=[_side_effect_question()]) return Step(replies=[_side_effect_question(ref)])
if reason == "quit": if reason == "quit":
# «Решил бросить» is the same signal as "брошу" in free text: rule `quit_intent` reacts (alert + # «Решил бросить» is the same signal as "брошу" in free text: rule `quit_intent` reacts (alert +
# delayed-onset explanation through the med_info action). # delayed-onset explanation through the med_info action).
@@ -301,53 +349,62 @@ def _skip_reason(reason: str) -> Any:
for _reason in SKIP_REASONS: 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") @med_trigger("side_effects")
async def _side_effects(ctx: Ctx, event: Event) -> Step: async def _side_effects(ctx: Ctx, event: Event, ref: str) -> Step:
return Step(replies=[_side_effect_question()]) return Step(replies=[_side_effect_question(ref)])
async def _after_side_effects(ctx: Ctx, codes: list[str]) -> Step: async def _after_side_effects(ctx: Ctx, codes: list[str], ref: str | None) -> Step:
await record_side_effects(ctx, codes) 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)] replies = [await side_effect_reply(ctx, codes)]
today = await checkin_of(ctx, ctx.now.date()) 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: if (today is None or today.med_taken is None) and await current_prescription(ctx) is not None:
t = load_meds_texts().side_effects 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")) 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: 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": if code == "other":
return Step( 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 return handler
for _code in SIDE_EFFECT_CODES: 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) @state(SIDE_EFFECT_TEXT_STATE)
async def _side_effect_text(ctx: Ctx, event: Event) -> Step: async def _side_effect_text(ctx: Ctx, event: Event) -> Step:
if event.type != "text" or ctx.text is None: 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) facts = await ctx.gateway.extract_checkin(ctx.text)
# Store only normalised codes, never the description itself (it stays in the scrubbed message log). # 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"] 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: def _dose_answer(taken: bool) -> Any:
async def handler(ctx: Ctx, event: Event) -> Step: async def handler(ctx: Ctx, event: Event, ref: str) -> Step:
await record_dose(ctx, taken, status=None if taken else "missed") await record_dose(ctx, taken, status=None if taken else "missed", ref=ref)
return Step( return Step(
replies=[Reply(text=load_meds_texts().taken_ack if taken else load_meds_texts().skip.ack["other"])], replies=[Reply(text=load_meds_texts().taken_ack if taken else load_meds_texts().skip.ack["other"])],
notify_monitoring=True, notify_monitoring=True,
@@ -356,8 +413,11 @@ def _dose_answer(taken: bool) -> Any:
return handler return handler
trigger("button:med:dose:yes")(_dose_answer(True)) med_trigger("dose:yes")(_dose_answer(True))
trigger("button:med:dose:no")(_dose_answer(False)) 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 # ------------------------------------------------------------------------------------------- question menu
+16 -31
View File
@@ -24,14 +24,14 @@ Data algorithm (medical data, jury requirement):
alcohol interaction, visit offer) come from the rules through `MonitoringService` (ADR-007). 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 * 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. 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` Also here: friendly intros for scheduled scales (pcl5, phq9, dar5, auditc), registered with
scales without an intro, so an exact `scale.<id>` enter hook adds one and delegates to the generic scale hook. `register_scale_intro` so the engine's `scale_due` / rule-triggered sessions show them.
""" """
import re import re
from dataclasses import replace
from datetime import datetime from datetime import datetime
from functools import lru_cache from functools import lru_cache
from pathlib import Path from pathlib import Path
@@ -59,9 +59,8 @@ from app.dialog.flows.meds import (
save_checkin, save_checkin,
side_effect_reply, side_effect_reply,
) )
from app.dialog.fsm import Ctx, Step, enter_hook_for, on_enter, state, trigger from app.dialog.fsm import Ctx, Step, abandonable, on_enter, state, trigger
from app.dialog.scale_session import PREFIX as SCALE_PREFIX from app.dialog.scale_session import register_scale_intro
from app.dialog.scale_session import active_scale, scale_state
PREFIX = "daily" PREFIX = "daily"
STEPS = ("mood", "med", "sleep", "nightmares", "alcohol", "text") 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) return _finish(ctx, replies)
abandonable(PREFIX)
@state(PREFIX) @state(PREFIX)
async def _handle(ctx: Ctx, event: Event) -> Step: async def _handle(ctx: Ctx, event: Event) -> Step:
step = ctx.state.split(".", 1)[1] if "." in ctx.state else "" 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 # ------------------------------------------------------------------------------------------- scale intros
async def _scale_intro(ctx: Ctx) -> Step: def _scale_intro(ctx: Ctx, scale_id: str, scheduled: bool, return_state: str) -> str | None:
"""Add the friendly intro of a scheduled (or rule-triggered) scale, then run the generic scale enter hook. """Friendly intro of a scheduled (or rule-triggered) scale. Onboarding sessions (#36) have their own texts."""
if ".onboarding" in return_state:
Onboarding sessions (#36) and sessions started with their own intro are left untouched. return None
""" intros = load_daily_texts().scale_intros
scale_id = active_scale(ctx.state) return (intros.scheduled if scheduled else intros.unscheduled).get(scale_id)
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()
for _scale_id in load_daily_texts().settings.intro_scales: 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)
+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):
+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:
# Текстовые команды отзыва согласия (сравниваются после приведения к нижнему регистру). # Текстовые команды отзыва согласия (сравниваются после приведения к нижнему регистру).
+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
+35 -30
View File
@@ -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] 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 def get_state() -> tuple[str, dict[str, Any]]:
async with session.get_clinical_session() as s: async with session.get_clinical_session() as s:
row = await repo.get_dialog_state(s, TOKEN) 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: async def test_reminder_without_dose_id_creates_dose_and_taken(engine: Engine, mon: ScriptedMonitoring) -> None:
r = await system(engine, "med_reminder") r = await system(engine, "med_reminder")
assert r[0].text == MEDS.reminder.text.format(drug="сертралин", dose="50 мг") assert r[0].text == MEDS.reminder.text.format(drug="сертралин", dose="50 мг")
assert buttons(r) == ["med:taken", "med:skip", "med:side_effects"]
[dose] = await doses() [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" 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 assert r[0].text == MEDS.taken_ack and mon.calls == 1
[dose] = await doses() [dose] = await doses()
assert dose.status == "taken" and dose.answered_at is not None 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: async def test_stale_reminder_button(engine: Engine) -> None:
await system(engine, "med_reminder") reminder = await system(engine, "med_reminder")
clock.advance(days=1) 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 assert r[0].text == MEDS.stale and await checkin() is None
@pytest.mark.parametrize("reason", ["forgot", "other"]) @pytest.mark.parametrize("reason", ["forgot", "other"])
async def test_skip_simple_reasons(engine: Engine, reason: str) -> None: async def test_skip_simple_reasons(engine: Engine, reason: str) -> None:
await system(engine, "med_reminder") r = await tap(engine, await system(engine, "med_reminder"), "med:skip")
r = await press(engine, "med:skip") assert kinds(r) == [f"med:skip_reason:{x}" for x in ("forgot", "side_effects", "quit", "other")]
assert buttons(r) == [f"med:skip_reason:{x}" for x in ("forgot", "side_effects", "quit", "other")] r = await tap(engine, r, f"med:skip_reason:{reason}")
r = await press(engine, f"med:skip_reason:{reason}")
assert r[0].text == MEDS.skip.ack[reason] assert r[0].text == MEDS.skip.ack[reason]
c = await checkin() c = await checkin()
assert c.med_taken is False and not c.quit_intent 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: async def test_skip_quit_sets_intent_and_explains(engine: Engine, mon: ScriptedMonitoring) -> None:
await system(engine, "med_reminder") skip = await tap(engine, await system(engine, "med_reminder"), "med:skip")
await press(engine, "med:skip")
mon.actions = [Action(type="med_info", code="quit_intent", params={"topic": "delayed_effect"})] 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 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 "2–4 недели" in r[1].text and MEDS.med_info["delayed_effect"].notice in r[1].text
assert (await checkin()).quit_intent is True assert (await checkin()).quit_intent is True
async def test_skip_because_of_side_effects_asks_which(engine: Engine) -> None: async def test_skip_because_of_side_effects_asks_which(engine: Engine) -> None:
await system(engine, "med_reminder") skip = await tap(engine, await system(engine, "med_reminder"), "med:skip")
await press(engine, "med:skip") r = await tap(engine, skip, "med:skip_reason:side_effects")
r = await press(engine, "med:skip_reason:side_effects") assert "med:se:nausea" in kinds(r) and "med:se:other" in kinds(r)
assert "med:se:nausea" in buttons(r) and "med:se:other" in buttons(r)
async def test_side_effect_button(engine: Engine, mon: ScriptedMonitoring) -> None: async def test_side_effect_button(engine: Engine, mon: ScriptedMonitoring) -> None:
await system(engine, "med_reminder") r = await tap(engine, await system(engine, "med_reminder"), "med:side_effects")
r = await press(engine, "med:side_effects") assert kinds(r) == [f"med:se:{c}" for c in ("nausea", "insomnia", "headache", "dizziness", "agitation", "other")]
assert buttons(r) == [f"med:se:{c}" for c in ("nausea", "insomnia", "headache", "dizziness", "agitation", "other")] r = await tap(engine, r, "med:se:nausea")
r = await press(engine, "med:se:nausea")
assert "проходит в течение 1–2 недель" in r[0].text # instruction excerpt 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 checkin()).side_effects == ["nausea"]
assert (await doses())[0].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" assert (await checkin()).med_taken is True and (await doses())[0].status == "taken"
async def test_side_effect_other_free_text(engine: Engine) -> None: async def test_side_effect_other_free_text(engine: Engine) -> None:
await system(engine, "med_reminder") reminder = await system(engine, "med_reminder")
await press(engine, "med:side_effects") r = await tap(engine, await tap(engine, reminder, "med:side_effects"), "med:se:other")
r = await press(engine, "med:se:other")
assert r[0].text == MEDS.side_effects.describe and (await get_state())[0] == "meds.side_effect_text" assert r[0].text == MEDS.side_effects.describe and (await get_state())[0] == "meds.side_effect_text"
r = await text(engine, "голова болит весь день") r = await text(engine, "голова болит весь день")
assert (await checkin()).side_effects == ["headache"] and (await get_state())[0] == "idle" assert (await checkin()).side_effects == ["headache"] and (await get_state())[0] == "idle"
await press(engine, "med:side_effects") await tap(engine, await tap(engine, reminder, "med:side_effects"), "med:se:other")
await press(engine, "med:se:other")
await text(engine, "какое-то странное ощущение") await text(engine, "какое-то странное ощущение")
assert (await checkin()).side_effects == ["headache", "other"] 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( await repo.upsert_checkin(
s, token=TOKEN, day=clock.today() - timedelta(days=1), now=clock.now(), side_effects=["nausea"] s, token=TOKEN, day=clock.today() - timedelta(days=1), now=clock.now(), side_effects=["nausea"]
) )
await system(engine, "med_reminder") se = await tap(engine, await system(engine, "med_reminder"), "med:side_effects")
r = await press(engine, "med:se:nausea") r = await tap(engine, se, "med:se:nausea")
assert r[0].text == MEDS.side_effects.repeat 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: async def test_reminder_during_checkin_keeps_checkin(engine: Engine) -> None:
await system(engine, "daily_checkin") await system(engine, "daily_checkin")
r = await system(engine, "med_reminder") 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 assert kinds(r[:1]) == ["med:taken", "med:skip", "med:side_effects"] and r[-1].text == DAILY.checkin.mood.question
r = await press(engine, "med:taken") r = await tap(engine, r, "med:taken")
assert r[0].text == MEDS.taken_ack and (await get_state())[0] == "daily.mood" assert r[0].text == MEDS.taken_ack and (await get_state())[0] == "daily.mood"
await press(engine, "daily:mood:5") await press(engine, "daily:mood:5")
assert (await get_state())[0] == "daily.sleep" # dose already answered → no med question assert (await get_state())[0] == "daily.sleep" # dose already answered → no med question
+9 -6
View File
@@ -292,13 +292,13 @@ async def test_andrey_13_days(andrey: RuleMonitoring) -> None:
dose_id = await new_dose(clock.today()) dose_id = await new_dose(clock.today())
reminder = await a.system("med_reminder", dose_id=dose_id, drug="сертралин", dose="50 мг") reminder = await a.system("med_reminder", dose_id=dose_id, drug="сертралин", dose="50 мг")
assert "сертралин 50 мг" in reminder[0].text 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: if taken:
r += await a.press("med:taken") r += await a.press(f"med:taken:{dose_id}")
else: else:
skip = await a.press("med:skip") skip = await a.press(f"med:skip:{dose_id}")
assert "med:skip_reason:quit" in buttons(skip) assert f"med:skip_reason:quit:{dose_id}" in buttons(skip)
r += skip + await a.press("med:skip_reason:quit") 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 # --- day 7: scheduled scales (PCL-5 = 44, PHQ-9 = 13, item 9 = 0) with friendly intros
if day == 7: 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 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) assert any(x.kind == "crisis" for x in r) and any(x.kind == "safety_plan" for x in r)
st, data = await state() 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"] acute = [x for x in await open_alerts() if x.level == "acute"]
assert [x.code for x in acute] == ["phq9_item9"] 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] assert patient_status(AlertLevel[x.level.upper()] for x in await open_alerts()).color == COLORS[12]