From fa40be3d941fa2abe05e2ff2dad53ed4ae3c5ceb Mon Sep 17 00:00:00 2001 From: Lacky1234union Date: Sat, 3 Oct 2026 15:42:35 +0300 Subject: [PATCH] =?UTF-8?q?feat(bot):=20dialog=20engine=20=E2=80=94=20scru?= =?UTF-8?q?b=20=E2=86=92=20safety=20=E2=86=92=20FSM=20=E2=86=92=20monitori?= =?UTF-8?q?ng,=20consent,=20age=20gate,=20crisis?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Engine.handle registered via register_engine; lexicon-first safety (no LLM on ACUTE), fixed crisis script once per open crisis, POSSIBLE → C-SSRS screen - FSM registry (@state/@on_enter/@trigger) with auto-loaded flows; core flow: new → consent → age_gate → module onboarding → idle, crisis, consent_revoked - reusable scale session (buttons, extract_item + clarify, item 9 / C-SSRS buttons-only, crisis flags), execution of monitoring dialog actions, texts in config/texts/dialog.yaml Refs #10 Co-Authored-By: Claude Opus 5.5 --- app/dialog/actions.py | 73 ++++++ app/dialog/crisis.py | 49 ++++ app/dialog/engine.py | 308 ++++++++++++++++++++++ app/dialog/flows/__init__.py | 16 ++ app/dialog/flows/core.py | 328 +++++++++++++++++++++++ app/dialog/fsm.py | 183 +++++++++++++ app/dialog/scale_session.py | 236 +++++++++++++++++ app/dialog/texts.py | 121 +++++++++ config/texts/dialog.yaml | 114 ++++++++ tests/test_dialog_engine.py | 485 +++++++++++++++++++++++++++++++++++ 10 files changed, 1913 insertions(+) create mode 100644 app/dialog/actions.py create mode 100644 app/dialog/crisis.py create mode 100644 app/dialog/engine.py create mode 100644 app/dialog/flows/__init__.py create mode 100644 app/dialog/flows/core.py create mode 100644 app/dialog/fsm.py create mode 100644 app/dialog/scale_session.py create mode 100644 app/dialog/texts.py create mode 100644 config/texts/dialog.yaml create mode 100644 tests/test_dialog_engine.py diff --git a/app/dialog/actions.py b/app/dialog/actions.py new file mode 100644 index 0000000..6143e37 --- /dev/null +++ b/app/dialog/actions.py @@ -0,0 +1,73 @@ +"""Execution of *dialog* actions returned by `MonitoringService.after_data_change()` (ADR-007). + +The rules (lane B, `config/rules.yaml`) decide WHAT happens; this module only turns an already chosen action +into replies. `alert` / `enhanced_monitoring` are handled by lane B itself (persisted + doctor notified), so they +are ignored here. While a crisis is open only crisis-relevant actions (crisis, safety plan, risk screen) reach +the patient: no medication tips or visit offers in the middle of a crisis. +""" + +from collections.abc import Iterable +from typing import Any + +from app.core.rules import Action +from app.db.clinical import SafetyPlan +from app.dialog.contracts import Button, Reply +from app.dialog.crisis import crisis_open, open_crisis +from app.dialog.fsm import Ctx, Step +from app.dialog.scale_session import active_scale, add_pending + +# med_info topic → phrase intent of `config/texts/phrases.yaml` (the LLM only rephrases the template). +_MED_INFO_INTENTS = {"delayed_effect": "quit_intent", "alcohol_interaction": "alcohol_info", "nausea": "nausea_info"} +_CRISIS_SAFE = {"crisis", "show_safety_plan", "schedule_scale"} + + +def _plan_line(step: Any) -> str: + if isinstance(step, dict): + return ": ".join(str(v) for v in step.values() if v) + return str(step) + + +async def safety_plan_reply(ctx: Ctx, block: str | None = None) -> Reply: + """The plan composed with the doctor (`safety_plans`), or the Stanley–Brown template from config.""" + row = await ctx.session.get(SafetyPlan, ctx.token) + steps = [_plan_line(s) for s in row.steps] if row is not None and row.steps else ctx.texts.safety_plan.template + lines = [ctx.texts.safety_plan.title, *(f"{i}. {s}" for i, s in enumerate(steps, 1))] + if block == "means_restriction": + lines += ["", ctx.texts.safety_plan.means_restriction] + return Reply(text="\n".join(lines), kind="safety_plan") + + +async def execute(ctx: Ctx, actions: Iterable[Action]) -> Step: + step = Step() + shown_plan = False + for action in actions: + if crisis_open({**ctx.data, **step.data}) and action.type not in _CRISIS_SAFE: + continue + if action.type == "crisis": + if not crisis_open({**ctx.data, **step.data}): + step = step.then(open_crisis(ctx, categories=[], source=f"rule:{action.code}")) + elif action.type == "show_safety_plan" and not shown_plan: + shown_plan = True + step = step.then(Step(replies=[await safety_plan_reply(ctx, action.params.get("block"))])) + elif action.type == "schedule_scale" and action.params.get("scale"): + scale_id = str(action.params["scale"]) + if active_scale(step.next_state or ctx.state) == scale_id: + continue # already being asked + data = {**ctx.data, **step.data} + pending = add_pending(data, scale_id, bool(action.params.get("scheduled", False))) + step = step.then(Step(data={"pending_scales": pending})) + elif action.type == "med_info": + intent = _MED_INFO_INTENTS.get(str(action.params.get("topic")), "default") + text = await ctx.gateway.phrase(intent, {}) + step = step.then(Step(replies=[Reply(text=text, kind="info")])) + elif action.type == "suggest_visit": + text = await ctx.gateway.phrase("book_visit_offer", {}) + button = Button(label=ctx.texts.visit.button, data="menu:visit") + step = step.then(Step(replies=[Reply(text=text, buttons=[[button]], kind="info")])) + elif action.type == "reach_out": + topic = str(action.params.get("topic", "default")) + texts = ctx.texts.reach_out + text = {"silence": texts.silence, "missed_doses": texts.missed_doses}.get(topic, texts.default) + step = step.then(Step(replies=[Reply(text=text)])) + # alert / enhanced_monitoring: lane B persists and notifies; nothing to say to the patient. + return step diff --git a/app/dialog/crisis.py b/app/dialog/crisis.py new file mode 100644 index 0000000..1b69f63 --- /dev/null +++ b/app/dialog/crisis.py @@ -0,0 +1,49 @@ +"""Crisis path of the dialog (guardrail #2): fixed texts only, never generated by the LLM. + +Decisions: + * The crisis script (`app.core.safety.crisis_reply`, `config/texts/crisis.yaml`) is sent verbatim, once per + open crisis. "Open" = `data["crisis"]["open"]` — set when the crisis starts, cleared only by the doctor's + acknowledgement (`doctor_ack`), so a patient repeating the phrase does not get the same long script again: + they get a short fixed "я рядом, врач уже уведомлён" + emergency resources instead. + * Entering the crisis drops unfinished scale sessions (their answered items are already stored item by item): + nothing else is asked while the patient is in crisis, except the C-SSRS screen. + * Without valid consent (before consent, after refusal/revocation, minors) we still answer a crisis message — + protecting life is the one processing ground that does not need consent (152-FZ art. 10 p. 2.3) — but with + a variant WITHOUT "your doctor has been notified", because no doctor is notified, and nothing is stored. +""" + +from typing import Any + +from app.core.safety import CrisisResource, CrisisScript, crisis_reply +from app.dialog.contracts import Reply +from app.dialog.fsm import Ctx, Step + + +def crisis_open(data: dict[str, Any]) -> bool: + return bool((data.get("crisis") or {}).get("open")) + + +def _resource_lines(resources: list[CrisisResource]) -> list[str]: + return [f"• {r.name}: {r.phone}" + (f" ({r.note})" if r.note else "") for r in resources] + + +def open_crisis(ctx: Ctx, categories: list[str], source: str) -> Step: + """Start the crisis (full script) or, if one is already open, send the short repeat text. + + `source` is a code for the dashboard/audit trail: "safety:lexicon", "safety:llm", "scale:phq9", "rule:". + """ + script = crisis_reply(categories) + if crisis_open(ctx.data): + text = "\n".join([ctx.texts.crisis.repeat, *_resource_lines(script.resources)]) + return Step(replies=[Reply(text=text, kind="crisis")]) + crisis = {"open": True, "opened_at": ctx.now.isoformat(), "source": source, "variant": script.variant} + return Step( + replies=[Reply(text=script.text, kind="crisis")], + next_state="crisis", + data={"crisis": crisis, "scales": None}, + ) + + +def no_consent_crisis_text(script: CrisisScript) -> str: + """Crisis script without the "doctor notified" line — for people we may not (or did not) notify about.""" + return "\n".join([script.body, "", *_resource_lines(script.resources), "", script.grounding]) diff --git a/app/dialog/engine.py b/app/dialog/engine.py new file mode 100644 index 0000000..6ddbdd2 --- /dev/null +++ b/app/dialog/engine.py @@ -0,0 +1,308 @@ +"""Dialog engine (issue #10, ADR-004): `Engine.handle(token, event) -> list[Reply]`, channel-agnostic. + +Pipeline for one incoming event (SPEC §3, guardrails #1, #2, #3): + + 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 + (protecting life does not need consent) and the core flow shows consent/refusal. + 3. safety.check() lexicon FIRST. Lexicon ACUTE → the LLM is not called at all. Otherwise the LLM risk + classifier is an auxiliary signal combined as max(lexicon, llm): it can only raise risk. + ACUTE → fixed crisis script verbatim (once per open crisis), state `crisis`, structured + `safety_events` fact for lane B, MonitoringService.after_data_change() (lane B alerts). + POSSIBLE → C-SSRS screen (scale session) and back to the previous state (or crisis if high). + 4. FSM the current state's handler (`app/dialog/flows/*`, `app/dialog/scale_session.py`). + button / system / start events skip 1–3 (no free text) and go through the same FSM. + + After the FSM step: if clinical data were written → commit → `after_data_change()` → the returned *dialog* + actions are executed (`app/dialog/actions.py`). The engine never creates alerts (ADR-007). + Then a queued scale starts if the FSM rests (idle; only the C-SSRS while in crisis). + Persist: inbound message (scrubbed) + outbound replies + `dialog_state` — only with active consent. + +Logs carry only the token, states and event codes (guardrail #8). +""" + +import logging +from dataclasses import dataclass, field +from datetime import datetime +from typing import Any + +from sqlalchemy.exc import InterfaceError, OperationalError +from sqlalchemy.ext.asyncio import AsyncSession + +from app.core import clock, safety +from app.core.privacy import scrub +from app.core.safety import SafetyResult, crisis_reply +from app.core.types import RiskLevel, ScrubbedText +from app.db import repo +from app.db.session import get_clinical_session +from app.dialog import scale_session +from app.dialog.actions import execute +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.texts import DialogTexts, load_dialog_texts +from app.llm.gateway import LLMGateway, get_gateway +from app.services.contracts import MonitoringService, get_monitoring + +log = logging.getLogger(__name__) + +# Enter hooks may chain (scale done → next onboarding state → next scale); a cycle must not hang a request. +MAX_ENTER_CHAIN = 8 +# DB unreachable: answer safely instead of failing the channel (connectivity errors only — bugs must surface). +_DB_DOWN = (OSError, OperationalError, InterfaceError) + + +@dataclass +class _Run: + replies: list[Reply] = field(default_factory=list) + notify: bool = False + + +def _consented(ctx: Ctx) -> bool: + return ctx.data.get("consent") is True and ctx.state not in NO_CONSENT_STATES + + +class Engine: + """Real `DialogEngine`. `gateway` / `monitoring` are injectable for tests; by default resolved per call so + a monitoring implementation registered later (lane B, #38) is picked up.""" + + def __init__(self, *, gateway: LLMGateway | None = None, monitoring: MonitoringService | None = None) -> None: + self._gateway = gateway + self._monitoring = monitoring + + async def handle(self, token: str, event: Event) -> list[Reply]: + texts = load_dialog_texts() + try: + async with get_clinical_session() as session: + return await self._handle(token, event, session, texts) + except _DB_DOWN as exc: + log.error("dialog db unavailable token=%s event=%s error=%s", token, event.type, type(exc).__name__) + return _degraded(event, texts) + + # ------------------------------------------------------------------------------------------ pipeline + + async def _handle(self, token: str, event: Event, session: AsyncSession, texts: DialogTexts) -> list[Reply]: + row = await repo.get_dialog_state(session, token) + ctx = Ctx( + token=token, + state=row.state if row is not None else "new", + data=dict(row.data or {}) if row is not None else {}, + now=clock.now(), + session=session, + gateway=self._gateway or get_gateway(), + monitoring=self._monitoring or get_monitoring(), + texts=texts, + ) + state_before = ctx.state + run = _Run() + + await self._apply(ctx, await self._route(ctx, event), run) + + if run.notify: + # Monitoring (lane B) reads the DB in its own session: commit our writes first. + await self._save_state(ctx) + await session.commit() + actions = await ctx.monitoring.after_data_change(token) + await self._apply(ctx, await execute(ctx, actions), run) + + await self._apply(ctx, await self._start_pending(ctx), run) + + if _consented(ctx): + for reply in run.replies: + # Outgoing texts may quote the safety plan (names of relatives) — scrub before storing too. + await repo.add_message( + session, token=token, direction="out", text=scrub(reply.text), ts=ctx.now, fsm_state=ctx.state + ) + await self._save_state(ctx) + log.info( + "dialog token=%s event=%s code=%s state=%s->%s replies=%d", + token, + event.type, + _event_code(event), + state_before, + ctx.state, + len(run.replies), + ) + return run.replies + + async def _route(self, ctx: Ctx, event: Event) -> Step: + consented = _consented(ctx) + if event.type == "text": + # (1) scrub before anything else: from here on only ScrubbedText exists for this message. + ctx.text = scrub(event.text or "") + if consented and _is_revoke_command(event.text or "", ctx.texts): + return await revoke(ctx) + if not consented: + return await self._no_consent_text(ctx, event) + await self._store_in(ctx, ctx.text) + return await self._safety(ctx, event) + + if event.type == "button": + if consented and event.button == "consent:revoke": + return await revoke(ctx) + if consented: + await self._store_in(ctx, scrub(f"[button] {event.button}")) # button data comes from the client + return await self._dispatch(ctx, event) + + if event.type == "system": + if event.system == "doctor_ack": + return close_crisis(ctx) + if not consented: + return Step() # nothing is pushed to people without consent + if event.system == "scale_due" and event.params.get("scale"): + scheduled = bool(event.params.get("scheduled", True)) + pending = scale_session.add_pending(ctx.data, str(event.params["scale"]), scheduled) + return Step(data={"pending_scales": pending}) # starts now if resting, else after the current step + return await self._dispatch(ctx, event) + + return await self._dispatch(ctx, event) # start + + async def _no_consent_text(self, ctx: Ctx, event: Event) -> Step: + assert ctx.text is not None + # Lexicon only: without consent the text must not reach the LLM (third-party processing). + result = safety.check(ctx.text) + if result.level == RiskLevel.ACUTE: + log.warning("safety acute without consent token=%s rules=%s", ctx.token, result.matched_rules) + return Step(replies=[Reply(text=no_consent_crisis_text(crisis_reply(result.categories)), kind="crisis")]) + return await self._dispatch(ctx, event) + + async def _safety(self, ctx: Ctx, event: Event) -> Step: + """Guardrail #2: the safety decision is made before any LLM sees the message for anything else.""" + assert ctx.text is not None + lexicon = safety.check(ctx.text) + if lexicon.level == RiskLevel.ACUTE: + result = lexicon # an acute message never reaches the LLM — not even the risk classifier + else: + result = safety.check(ctx.text, llm_signal=await ctx.gateway.risk_classify(ctx.text)) + + if result.level == RiskLevel.ACUTE: + handled = "crisis_repeat" if crisis_open(ctx.data) else "crisis_script" + log.warning("safety acute token=%s rules=%s handled=%s", ctx.token, result.matched_rules, handled) + step = open_crisis(ctx, result.categories, source=f"safety:{result.source}") + step.data["safety_events"] = self._safety_events(ctx, result, handled) + step.notify_monitoring = True # lane B escalates to the doctor on call + return step + + if result.level == RiskLevel.POSSIBLE: + screen = ctx.texts.settings.risk_screen_scale + if self._can_screen(ctx, screen): + step = await scale_session.start_scale( + ctx, screen, return_state=ctx.state, scheduled=False, intro=ctx.texts.scale.risk_screen_intro + ) + step.data["risk_screen_at"] = ctx.now.isoformat() + handled = "risk_screen" + else: + step = await self._dispatch(ctx, event) + handled = "dispatched" + log.info("safety possible token=%s rules=%s handled=%s", ctx.token, result.matched_rules, handled) + step.data["safety_events"] = self._safety_events(ctx, result, handled) + step.notify_monitoring = True + return step + + return await self._dispatch(ctx, event) + + def _can_screen(self, ctx: Ctx, screen: str) -> bool: + if crisis_open(ctx.data) or scale_session.active_scale(ctx.state) == screen: + return False + last = ctx.data.get("risk_screen_at") + if last is None: + return True + hours = (ctx.now - datetime.fromisoformat(last)).total_seconds() / 3600 + return hours >= ctx.texts.settings.cssrs_cooldown_hours + + def _safety_events(self, ctx: Ctx, result: SafetyResult, handled: str) -> list[dict[str, Any]]: + """Structured safety fact for lane B (#38): no text, only codes. Newest last, capped.""" + event = { + "ts": ctx.now.isoformat(), + "level": result.level.name.lower(), + "categories": result.categories, + "rules": result.matched_rules, + "source": result.source, + "handled": handled, + "state": ctx.state, + } + events = [*(ctx.data.get("safety_events") or []), event] + return events[-ctx.texts.settings.max_safety_events :] + + async def _dispatch(self, ctx: Ctx, event: Event) -> Step: + handler = handler_for(ctx.state) + if handler is None: + # A state from a removed/renamed flow: recover instead of getting stuck. + log.error("dialog unknown state token=%s state=%s", ctx.token, ctx.state) + return Step(next_state="idle" if _consented(ctx) else "new") + return await handler(ctx, event) + + async def _apply(self, ctx: Ctx, step: Step, run: _Run, depth: int = 0) -> None: + run.replies.extend(step.replies) + apply_data(ctx.data, step.data) + run.notify = run.notify or step.notify_monitoring + if not step.next_state or step.next_state == ctx.state: + return + ctx.state = step.next_state + hook = enter_hook_for(ctx.state) + if hook is None: + return + if depth >= MAX_ENTER_CHAIN: + log.error("dialog enter chain too long token=%s state=%s", ctx.token, ctx.state) + return + await self._apply(ctx, await hook(ctx), run, depth + 1) + + async def _start_pending(self, ctx: Ctx) -> Step: + """Start a queued scale when nothing else is being asked. In crisis only the risk screen may run.""" + pending = list(ctx.data.get("pending_scales") or []) + if not pending or ctx.state not in RESTING_STATES: + return Step() + screen = ctx.texts.settings.risk_screen_scale + if ctx.state == "crisis": + pending_screen = [p for p in pending if p["scale"] == screen] + if not pending_screen: + return Step() + item = pending_screen[0] + else: + item = pending[0] + rest = [p for p in pending if p is not item] + step = await scale_session.start_scale( + ctx, + item["scale"], + return_state=ctx.state, + scheduled=bool(item.get("scheduled", True)), + intro=ctx.texts.scale.risk_screen_intro if item["scale"] == screen else None, + ) + step.data["pending_scales"] = rest or None + return step + + async def _store_in(self, ctx: Ctx, text: ScrubbedText) -> None: + await repo.add_message(ctx.session, token=ctx.token, direction="in", text=text, ts=ctx.now, fsm_state=ctx.state) + + async def _save_state(self, ctx: Ctx) -> None: + await repo.set_dialog_state(ctx.session, token=ctx.token, state=ctx.state, data=ctx.data, now=ctx.now) + + +def _is_revoke_command(raw: str, texts: DialogTexts) -> bool: + return raw.strip().lower().rstrip(".!") in texts.commands.revoke + + +def _event_code(event: Event) -> str: + if event.type == "button": + return event.button or "" + if event.type == "system": + return event.system or "" + return "-" + + +def _degraded(event: Event, texts: DialogTexts) -> list[Reply]: + """DB down: still answer a crisis message (lexicon only, no "doctor notified" — nobody could be notified).""" + if event.type == "text" and event.text: + result = safety.check(scrub(event.text)) + if result.level == RiskLevel.ACUTE: + return [Reply(text=no_consent_crisis_text(crisis_reply(result.categories)), kind="crisis")] + if event.type == "system": + return [] + return [Reply(text=texts.unavailable, kind="info")] + + +load_flows() +register_engine(Engine) diff --git a/app/dialog/flows/__init__.py b/app/dialog/flows/__init__.py new file mode 100644 index 0000000..42613d4 --- /dev/null +++ b/app/dialog/flows/__init__.py @@ -0,0 +1,16 @@ +"""Dialog flows. Each module here registers its FSM states at import time (`@state`, `@on_enter`, `@trigger`). + +A new direction (veteran #36/#37, family, depression) is a new file `app/dialog/flows/.py`; it is picked +up automatically by `load_flows()` — the engine is not edited. +""" + +import pkgutil +from importlib import import_module + + +def load_flows() -> None: + """Import every flow module of this package (core first, so flows may override core triggers).""" + import_module(f"{__name__}.core") + for info in sorted(pkgutil.iter_modules(__path__), key=lambda m: m.name): + if info.name != "core" and not info.name.startswith("_"): + import_module(f"{__name__}.{info.name}") diff --git a/app/dialog/flows/core.py b/app/dialog/flows/core.py new file mode 100644 index 0000000..5f4e09a --- /dev/null +++ b/app/dialog/flows/core.py @@ -0,0 +1,328 @@ +"""Core flow shared by all modules: connect → consent → age gate → module onboarding → idle; crisis; revocation. + + new ──start(code)──▶ consent ──«Согласен»──▶ age_gate ──adult──▶ .onboarding.start … ──▶ idle + │ └──< 18──▶ minor (stop, child helpline) + └──«Не согласен»──▶ refused (stop, /start to reconsider) + any consented state ──«Отозвать согласие»──▶ consent_revoked (silent; /start → consent again) + any ──ACUTE / item 9 / C-SSRS high──▶ crisis ──doctor_ack──▶ idle + +Consent and age are the registration step, so this module (and only it) touches the identity DB, through +`app.db.identity_repo`, and only for consent / revocation / birth year (guardrail #5). +""" + +import re + +from app.db import identity_repo, repo +from app.db.session import get_identity_session +from app.dialog.actions import safety_plan_reply +from app.dialog.contracts import Button, Event, Reply +from app.dialog.fsm import Ctx, Step, get_trigger, is_registered, on_enter, state, trigger +from app.integrations.base import integrations +from app.services.contracts import NoopMonitoring + +ADULT_AGE = 18 +_YEAR_RE = re.compile(r"\b(19\d{2}|20\d{2})\b") +_AGE_RE = re.compile(r"^\s*(\d{1,3})\s*(лет|год|года)?\s*$") + + +# ------------------------------------------------------------------------------------------- replies + + +def consent_reply(ctx: Ctx) -> Reply: + t = ctx.texts.consent + buttons = [[Button(label=t.accept, data="consent:yes")], [Button(label=t.decline, data="consent:no")]] + return Reply(text=t.text, buttons=buttons, kind="question") + + +def menu_reply(ctx: Ctx, text: str | None = None) -> Reply: + b = ctx.texts.idle.buttons + buttons = [ + [Button(label=b.checkin, data="menu:checkin")], + [Button(label=b.med_question, data="menu:med_question")], + [Button(label=b.contact_doctor, data="menu:contact_doctor")], + [Button(label=b.revoke, data="consent:revoke")], + ] + return Reply(text=text or ctx.texts.idle.menu_prompt, buttons=buttons) + + +# ------------------------------------------------------------------------------------------- onboarding API + + +async def onboarding_entry(ctx: Ctx, after: str | None = None) -> str: + """Entry state of the next module onboarding: `"{module}.onboarding.start"` for the first module of the + profile (after `after`, if given) that has such a state registered; `"idle"` when none is left.""" + profile = await ctx.profile() + modules = list(profile.modules) if profile is not None and profile.modules else [] + if after in modules: + modules = modules[modules.index(after) + 1 :] + for module in modules: + if is_registered(f"{module}.onboarding.start"): + return f"{module}.onboarding.start" + return "idle" + + +async def finish_onboarding(ctx: Ctx, module: str | None = None) -> Step: + """A module flow calls this when its onboarding is done: next module's onboarding, or idle + welcome.""" + nxt = await onboarding_entry(ctx, after=module) + if nxt == "idle": + return Step(replies=[menu_reply(ctx, ctx.texts.idle.welcome)], next_state="idle") + return Step(next_state=nxt) + + +# ------------------------------------------------------------------------------------------- new / consent + + +@state("new") +async def _new(ctx: Ctx, event: Event) -> Step: + code = event.params.get("code") if event.type == "start" else None + if code: + # Telegram deep link carries the doctor's one-time connect code; the web console passes the token + # directly (no code) — then the token is already known to the system. + person = await integrations.identity.verify(str(code)) + if person is None or person.token != ctx.token: + return Step(replies=[Reply(text=ctx.texts.consent.code_invalid)]) + if await ctx.profile() is None: + await repo.upsert_profile(ctx.session, token=ctx.token, doctor_id=person.doctor_id) + return Step(replies=[consent_reply(ctx)], next_state="consent") + + +@state("consent") +async def _consent(ctx: Ctx, event: Event) -> Step: + if event.type == "button" and event.button == "consent:yes": + version = ctx.texts.consent.version + async with get_identity_session() as s: + # None when the token has no identity record (web demo console) — the dialog state keeps the fact. + await identity_repo.record_consent(s, token=ctx.token, version=version, now=ctx.now) + data = {"consent": True, "consent_ver": version, "consent_at": ctx.now.isoformat(), "revoked_at": None} + return Step(next_state="age_gate", data=data) + if event.type == "button" and event.button == "consent:no": + return _refused_step(ctx) + # Anything else (incl. free text) is not processed before consent: show the consent text again. + return Step(replies=[consent_reply(ctx)]) + + +def _refused_step(ctx: Ctx) -> Step: + button = Button(label=ctx.texts.consent.reconsider, data="consent:restart") + return Step(replies=[Reply(text=ctx.texts.consent.refused, buttons=[[button]])], next_state="refused") + + +@state("refused") +async def _refused(ctx: Ctx, event: Event) -> Step: + if event.type == "start" or (event.type == "button" and event.button == "consent:restart"): + return Step(replies=[consent_reply(ctx)], next_state="consent") + if event.type == "system": + return Step() + return _refused_step(ctx) + + +# ------------------------------------------------------------------------------------------- age gate + + +def _age_verdict(birth_year: int, current_year: int) -> str: + """Without the birthday the age is `current - birth_year` or one less; ask when the two disagree on 18.""" + oldest, youngest = current_year - birth_year, current_year - birth_year - 1 + if youngest >= ADULT_AGE: + return "adult" + if oldest < ADULT_AGE: + return "minor" + return "unsure" + + +def _age_question(ctx: Ctx, note: str = "") -> Reply: + t = ctx.texts.age_gate + buttons = [[Button(label=t.adult, data="age:adult")], [Button(label=t.minor, data="age:minor")]] + return Reply(text="\n\n".join(p for p in (note, t.question) if p), buttons=buttons, kind="question") + + +@on_enter("age_gate") +async def _age_enter(ctx: Ctx) -> Step: + profile = await ctx.profile() + birth_year = profile.birth_year if profile is not None else None + if birth_year is None: + async with get_identity_session() as s: + patient = await identity_repo.get_patient(s, ctx.token) + birth_year = patient.birth_year if patient is not None else None + if birth_year is not None: + verdict = _age_verdict(birth_year, ctx.now.year) + if verdict == "adult": + return await _adult(ctx) + if verdict == "minor": + return await _minor(ctx) + return Step(replies=[_age_question(ctx)]) + + +@state("age_gate") +async def _age_gate(ctx: Ctx, event: Event) -> Step: + if event.type == "button" and event.button == "age:adult": + return await _adult(ctx) + if event.type == "button" and event.button == "age:minor": + return await _minor(ctx) + if event.type == "text" and ctx.text: + year = _YEAR_RE.search(ctx.text) + age_match = _AGE_RE.match(ctx.text) + if year: + birth_year = int(year.group(1)) + verdict = _age_verdict(birth_year, ctx.now.year) + if verdict == "adult": + return await _adult(ctx, birth_year) + if verdict == "minor": + return await _minor(ctx) + elif age_match: + age = int(age_match.group(1)) + return await (_adult(ctx) if age >= ADULT_AGE else _minor(ctx)) + return Step(replies=[_age_question(ctx, ctx.texts.age_gate.unclear)]) + return Step(replies=[_age_question(ctx)]) + + +async def _adult(ctx: Ctx, birth_year: int | None = None) -> Step: + if birth_year is not None and await ctx.profile() is not None: + await repo.upsert_profile(ctx.session, token=ctx.token, birth_year=birth_year) + return Step(data={"adult": True}).then(await finish_onboarding(ctx)) + + +async def _minor(ctx: Ctx) -> Step: + # A minor cannot give this consent on their own: withdraw it and stop processing. + async with get_identity_session() as s: + await identity_repo.revoke_consent(s, token=ctx.token, now=ctx.now) + return Step( + replies=[Reply(text=ctx.texts.age_gate.minor_stop, kind="info")], + next_state="minor", + data={"consent": False, "minor": True}, + ) + + +@state("minor") +async def _minor_state(ctx: Ctx, event: Event) -> Step: + if event.type == "system": + return Step() + return Step(replies=[Reply(text=ctx.texts.age_gate.minor_stop, kind="info")]) + + +# ------------------------------------------------------------------------------------------- idle + + +@state("idle") +async def _idle(ctx: Ctx, event: Event) -> Step: + if event.type == "start": + return Step(replies=[menu_reply(ctx)]) + if event.type == "button" and event.button: + handler = get_trigger(f"button:{event.button}") + return await handler(ctx, event) if handler else Step(replies=[menu_reply(ctx)]) + if event.type == "system" and event.system: + handler = get_trigger(f"system:{event.system}") + return await handler(ctx, event) if handler else Step() + if event.type == "text": + handler = get_trigger("text:idle") + if handler: + return await handler(ctx, event) + # Neutral empathic reply; the LLM only rephrases the approved template (and is filtered). + return Step(replies=[menu_reply(ctx, await ctx.gateway.phrase("support"))]) + return Step() + + +# Default idle triggers — flows (#15, #36, #37) override them with `@trigger(..., replace=True)`. + + +@trigger("button:menu:checkin") +async def _menu_checkin(ctx: Ctx, event: Event) -> Step: + return Step(replies=[Reply(text=ctx.texts.idle.stub_checkin)]) + + +@trigger("button:menu:med_question") +async def _menu_med_question(ctx: Ctx, event: Event) -> Step: + return Step(replies=[Reply(text=ctx.texts.idle.stub_med_question)]) + + +@trigger("button:menu:contact_doctor") +async def _menu_contact_doctor(ctx: Ctx, event: Event) -> Step: + # A fact for lane B (#38): the patient asked the doctor to get in touch. No alert is created here (ADR-007). + requests = [*(ctx.data.get("contact_requests") or []), {"ts": ctx.now.isoformat()}][-20:] + return Step( + replies=[Reply(text=ctx.texts.idle.contact_doctor_ack)], + data={"contact_requests": requests}, + notify_monitoring=True, + ) + + +@trigger("button:menu:visit") +async def _menu_visit(ctx: Ctx, event: Event) -> Step: + return Step(replies=[Reply(text=ctx.texts.idle.visit_stub)]) + + +@trigger("button:safety_plan") +async def _show_plan(ctx: Ctx, event: Event) -> Step: + return Step(replies=[await safety_plan_reply(ctx)]) + + +@trigger("system:med_reminder") +async def _med_reminder(ctx: Ctx, event: Event) -> Step: + drug = str(event.params.get("drug", "")) + return Step(replies=[Reply(text=await ctx.gateway.phrase("med_reminder", {"drug": drug}))]) + + +@trigger("system:daily_checkin") +async def _daily_checkin(ctx: Ctx, event: Event) -> Step: + return Step(replies=[menu_reply(ctx, ctx.texts.idle.stub_checkin)]) + + +@trigger("system:reach_out") +async def _reach_out(ctx: Ctx, event: Event) -> Step: + return Step(replies=[Reply(text=ctx.texts.reach_out.default)]) + + +# ------------------------------------------------------------------------------------------- crisis + + +def close_crisis(ctx: Ctx) -> Step: + """Doctor acknowledged the crisis: clear the flag; a C-SSRS still running returns to idle afterwards.""" + crisis = {**(ctx.data.get("crisis") or {}), "open": False, "closed_at": ctx.now.isoformat()} + scales = { + sid: {**s, "return_state": "idle"} if s.get("return_state") == "crisis" else s + for sid, s in (ctx.data.get("scales") or {}).items() + } + step = Step(data={"crisis": crisis, "scales": scales}) + if ctx.state == "crisis": + step.next_state = "idle" + step.replies.append(Reply(text=ctx.texts.crisis.closed)) + return step + + +@state("crisis") +async def _crisis(ctx: Ctx, event: Event) -> Step: + # The NoopMonitoring stub knows nothing about crises and always says "not in crisis"; trusting it would + # end every crisis on the next message. Until lane B (#38) lands only `doctor_ack` ends a crisis. + if not isinstance(ctx.monitoring, NoopMonitoring) and not await ctx.monitoring.is_in_crisis(ctx.token): + return close_crisis(ctx) + if event.type == "system": + return Step() # reminders and check-ins wait until the crisis is over + if event.type == "button" and event.button == "safety_plan": + return Step(replies=[await safety_plan_reply(ctx)]) + plan = Button(label=ctx.texts.safety_plan.title.rstrip(":"), data="safety_plan") + return Step(replies=[Reply(text=ctx.texts.crisis.hold, buttons=[[plan]], kind="crisis")]) + + +# ------------------------------------------------------------------------------------------- revocation + + +async def revoke(ctx: Ctx) -> Step: + """Withdraw consent in one step (152-FZ art. 9 p. 2): identity DB + dialog state; the bot goes silent. + + Clinical data are marked for deletion (`data["delete_requested"]`) — erasure itself is lane B's job. + """ + async with get_identity_session() as s: + await identity_repo.revoke_consent(s, token=ctx.token, now=ctx.now) + data = { + "consent": False, + "revoked_at": ctx.now.isoformat(), + "delete_requested": True, + "scales": None, + "pending_scales": None, + } + return Step(replies=[Reply(text=ctx.texts.revoked.confirm)], next_state="consent_revoked", data=data) + + +@state("consent_revoked") +async def _revoked(ctx: Ctx, event: Event) -> Step: + if event.type == "start": + return Step(replies=[consent_reply(ctx)], next_state="consent") + return Step() # silent: nothing is processed after revocation diff --git a/app/dialog/fsm.py b/app/dialog/fsm.py new file mode 100644 index 0000000..e1b3860 --- /dev/null +++ b/app/dialog/fsm.py @@ -0,0 +1,183 @@ +"""FSM framework of the dialog engine: context, step result and the state / trigger registries. + +States are plain strings, namespaced by module ("veteran.onboarding.pcl5"); core states have short names +("new", "consent", "age_gate", "idle", "crisis", "consent_revoked", ...). A flow module +(`app/dialog/flows/.py`) registers its states at import time — the engine never has to be edited to +add a scenario: + + from app.dialog.fsm import Ctx, Step, on_enter, state + from app.dialog.scale_session import start_scale + + @on_enter("veteran.onboarding.start") # runs when the FSM *enters* the state + async def _start(ctx: Ctx) -> Step: + return await start_scale(ctx, "pcptsd5", return_state="veteran.onboarding.pcl5") + + @state("veteran.onboarding.trauma") # handles events while the FSM *is in* the state + async def _trauma(ctx: Ctx, event: Event) -> Step: + ... + return Step(replies=[...], next_state="idle", data={"trauma": True}) + +Name resolution: an exact registration wins, otherwise the longest registered dotted prefix handles the state +("scale.phq9" → the generic "scale" handler). Triggers (`@trigger("button:menu:checkin")`, +`@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`. +""" + +from __future__ import annotations + +from collections.abc import Awaitable, Callable +from dataclasses import dataclass, field +from datetime import datetime +from typing import Any + +from sqlalchemy.ext.asyncio import AsyncSession + +from app.core.types import ScrubbedText +from app.db import repo +from app.db.clinical import PatientProfile +from app.dialog.contracts import Event, Reply +from app.dialog.texts import DialogTexts +from app.llm.gateway import LLMGateway +from app.services.contracts import MonitoringService + +# States where nothing is being asked: a pending scale may start here (see engine `_start_pending`). +RESTING_STATES = frozenset({"idle", "crisis"}) +# States without (valid) consent: the engine stores nothing and never calls the LLM here (152-FZ art. 9-10). +NO_CONSENT_STATES = frozenset({"new", "consent", "refused", "minor", "consent_revoked"}) + + +@dataclass +class Ctx: + """Everything a state handler may use. `data` is the persisted `dialog_state.data` (treat as read-only: + return changes in `Step.data`, the engine applies them).""" + + token: str + state: str + data: dict[str, Any] + now: datetime # from app.core.clock, taken once per incoming event (ADR-005) + session: AsyncSession # clinical DB unit of work of this event + gateway: LLMGateway + monitoring: MonitoringService + texts: DialogTexts + text: ScrubbedText | None = None # scrubbed patient text for text events — the only text handlers see + _profile: PatientProfile | None = field(default=None, repr=False) + _profile_loaded: bool = field(default=False, repr=False) + + async def profile(self) -> PatientProfile | None: + if not self._profile_loaded: + self._profile = await repo.get_profile(self.session, self.token) + self._profile_loaded = True + return self._profile + + +@dataclass +class Step: + """Result of a handler. + + replies — messages to send, in order; + next_state — new FSM state (None = stay); entering a different state runs its `@on_enter` hook; + data — updates merged into `dialog_state.data` (a value of None deletes the key); + notify_monitoring — the handler wrote clinical data: the engine commits and calls + `MonitoringService.after_data_change()`, then executes the returned dialog actions. + """ + + replies: list[Reply] = field(default_factory=list) + next_state: str | None = None + data: dict[str, Any] = field(default_factory=dict) + notify_monitoring: bool = False + + def then(self, other: Step) -> Step: + """Sequential composition: replies concatenated, later state/data win.""" + return Step( + replies=[*self.replies, *other.replies], + next_state=other.next_state or self.next_state, + data={**self.data, **other.data}, + notify_monitoring=self.notify_monitoring or other.notify_monitoring, + ) + + +Handler = Callable[[Ctx, Event], Awaitable[Step]] +EnterHook = Callable[[Ctx], Awaitable[Step]] + +_handlers: dict[str, Handler] = {} +_enter_hooks: dict[str, EnterHook] = {} +_triggers: dict[str, Handler] = {} + + +def state(name: str) -> Callable[[Handler], Handler]: + """Register the event handler of state `name` (or of a dotted prefix, e.g. "scale").""" + + def deco(fn: Handler) -> Handler: + if name in _handlers: + raise ValueError(f"state '{name}' already registered") + _handlers[name] = fn + return fn + + return deco + + +def on_enter(name: str) -> Callable[[EnterHook], EnterHook]: + """Register a hook that runs when the FSM enters state `name` (asks the first question, chains flows).""" + + def deco(fn: EnterHook) -> EnterHook: + if name in _enter_hooks: + raise ValueError(f"enter hook of '{name}' already registered") + _enter_hooks[name] = fn + return fn + + return deco + + +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. + """ + + def deco(fn: Handler) -> Handler: + if key in _triggers and not replace: + raise ValueError(f"trigger '{key}' already registered (use replace=True to override)") + _triggers[key] = fn + return fn + + return deco + + +def _resolve(name: str, table: dict[str, Any]) -> str | None: + candidate = name + while candidate: + if candidate in table: + return candidate + if "." not in candidate: + return None + candidate = candidate.rsplit(".", 1)[0] + return None + + +def handler_for(name: str) -> Handler | None: + key = _resolve(name, _handlers) + return _handlers[key] if key else None + + +def enter_hook_for(name: str) -> EnterHook | None: + key = _resolve(name, _enter_hooks) + return _enter_hooks[key] if key else None + + +def get_trigger(key: str) -> Handler | None: + return _triggers.get(key) + + +def is_registered(name: str) -> bool: + """Exact registration (handler or enter hook) — used to find a module's onboarding entry point.""" + return name in _handlers or name in _enter_hooks + + +def apply_data(data: dict[str, Any], updates: dict[str, Any]) -> None: + for key, value in updates.items(): + if value is None: + data.pop(key, None) + else: + data[key] = value diff --git a/app/dialog/scale_session.py b/app/dialog/scale_session.py new file mode 100644 index 0000000..366eded --- /dev/null +++ b/app/dialog/scale_session.py @@ -0,0 +1,236 @@ +"""Reusable scale session: asks a questionnaire item by item inside the dialog (SPEC §6 "Сбор шкалы в диалоге"). + +Usage from a flow (#36/#37): + + step = await start_scale(ctx, "pcl5", return_state="veteran.onboarding.phq9", scheduled=True) + return step # FSM goes to "scale.pcl5"; its enter hook asks item 1; on completion → return_state + +Algorithm per item: + 1. The item is shown with one button per answer option from `config/scales/.yaml`. + 2. A button press is taken as is (source="button"). + 3. Free text is accepted only for items without `buttons_only` (PHQ-9 item 9 and the whole C-SSRS are + buttons-only: suicidal-ideation answers are never interpreted by the LLM — SPEC §6, ADR-002). The scrubbed + text goes to `gateway.extract_item`; `None` (failure / confidence < 0.5 / out-of-range score) → a clarifying + question with the same buttons, nothing is stored (source="llm" only for a confident extraction). + 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. + 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. +""" + +from typing import Any + +from app.core.privacy import scrub +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 + +PREFIX = "scale" + + +def scale_state(scale_id: str) -> str: + return f"{PREFIX}.{scale_id}" + + +def active_scale(state_name: str) -> str | None: + return state_name.split(".", 1)[1] if state_name.startswith(PREFIX + ".") else None + + +def add_pending(data: dict[str, Any], scale_id: str, scheduled: bool) -> list[dict[str, Any]]: + """Queue a scale to start at the next resting state (idle/crisis); duplicates are ignored.""" + pending = list(data.get("pending_scales") or []) + if scale_id not in {p["scale"] for p in pending}: + pending.append({"scale": scale_id, "scheduled": scheduled}) + return pending + + +async def start_scale( + ctx: Ctx, scale_id: str, *, return_state: str, scheduled: bool = True, intro: str | None = None +) -> Step: + """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. + """ + scale = load_scale(scale_id) # fail fast on an unknown id + sessions = dict(ctx.data.get("scales") or {}) + sessions[scale_id] = { + "idx": 0, + "answers": {}, + "return_state": return_state, + "scheduled": scheduled, + "intro": intro, + "intro_shown": False, + } + target = scale_state(scale_id) + if ctx.state == target: # restart in place: the enter hook would not run + sessions[scale_id]["intro_shown"] = True + return Step(replies=[_question(ctx, scale, sessions[scale_id], intro=True)], data={"scales": sessions}) + return Step(next_state=target, data={"scales": sessions}) + + +def _question(ctx: Ctx, scale: ScaleDef, session: dict[str, Any], *, intro: bool = False, note: str = "") -> Reply: + idx = session["idx"] + item = scale.items[idx] + parts: list[str] = [] + if note: + parts.append(note) + if intro: + if session.get("intro"): + parts.append(session["intro"]) + parts.append(ctx.texts.scale.intro.format(count=len(scale.items), instruction=scale.instruction or "").strip()) + parts.append(ctx.texts.scale.question.format(n=idx + 1, total=len(scale.items), text=item.text)) + buttons = [[Button(label=o.label, data=f"{PREFIX}:{scale.id}:{item.id}:{o.value}")] for o in item.options] + return Reply(text="\n\n".join(parts), buttons=buttons, kind="question") + + +def _session(ctx: Ctx) -> tuple[ScaleDef, dict[str, Any]] | None: + scale_id = active_scale(ctx.state) + session = (ctx.data.get("scales") or {}).get(scale_id) if scale_id else None + if scale_id is None or session is None: + return None + return load_scale(scale_id), session + + +@on_enter(PREFIX) +async def _enter(ctx: Ctx) -> Step: + found = _session(ctx) + if found is None: + return Step(next_state="idle") + scale, session = found + show_intro = session["idx"] == 0 and not session.get("intro_shown") + updated = {**session, "intro_shown": True} + sessions = {**(ctx.data.get("scales") or {}), scale.id: updated} + return Step(replies=[_question(ctx, scale, updated, intro=show_intro)], data={"scales": sessions}) + + +def _parse_button(data: str, scale: ScaleDef, item: Item) -> int | None: + parts = data.split(":") + if len(parts) != 4 or parts[0] != PREFIX or parts[1] != scale.id or parts[2] != item.id: + return None # stale button from an earlier question + try: + value = int(parts[3]) + except ValueError: + return None + return value if value in {o.value for o in item.options} else None + + +@state(PREFIX) +async def _handle(ctx: Ctx, event: Event) -> Step: + found = _session(ctx) + if found is None: + return Step(next_state="idle") + scale, session = found + item = scale.items[session["idx"]] + + if event.type == "button" and event.button: + value = _parse_button(event.button, scale, item) + if value is not None: + return await _record(ctx, scale, session, item, value, source="button") + return Step(replies=[_question(ctx, scale, session)]) + + if event.type == "text" and ctx.text is not None: + if item.buttons_only: + # Safety items are never LLM-interpreted: ask again for a button. + return Step(replies=[_question(ctx, scale, session, note=ctx.texts.scale.buttons_only)]) + extracted = await ctx.gateway.extract_item(item.id, item.text, item.max_value, ctx.text) + if extracted is None or extracted.score not in {o.value for o in item.options}: + return Step(replies=[_question(ctx, scale, session, note=ctx.texts.scale.clarify)]) + return await _record( + ctx, + scale, + session, + item, + extracted.score, + source="llm", + confidence=extracted.confidence, + evidence=extracted.evidence, + ) + + if event.type == "system": + return Step() # reminders etc. do not interrupt a questionnaire + return Step(replies=[_question(ctx, scale, session)]) + + +async def _record( + ctx: Ctx, + scale: ScaleDef, + session: dict[str, Any], + item: Item, + value: int, + *, + source: str, + confidence: float | None = None, + evidence: str = "", +) -> Step: + await repo.add_item_response( + ctx.session, + token=ctx.token, + scale=scale.id, + item_id=item.id, + score=value, + source=source, + ts=ctx.now, + confidence=confidence, + # Evidence is a quote of already scrubbed text; scrub again so nothing unscrubbed can be stored. + evidence=scrub(evidence) if evidence else None, + ) + answers = {**session["answers"], item.id: value} + profile = await ctx.profile() + sex = profile.sex if profile is not None and profile.sex in ("m", "f") else None + result = score(scale.id, answers, sex=sex) # type: ignore[arg-type] + crisis_flags = [f for f in result.flags if f in ctx.texts.settings.crisis_flags] + if crisis_flags or session["idx"] + 1 >= len(scale.items): + return await _finish(ctx, scale, session, answers, result, crisis_flags) + updated = {**session, "answers": answers, "idx": session["idx"] + 1} + sessions = {**(ctx.data.get("scales") or {}), scale.id: updated} + return Step(replies=[_question(ctx, scale, updated)], data={"scales": sessions}) + + +async def _finish( + ctx: Ctx, + scale: ScaleDef, + session: dict[str, Any], + answers: dict[str, int], + result: ScaleResult, + crisis_flags: list[str], +) -> Step: + await repo.add_scale_result( + ctx.session, + token=ctx.token, + scale=scale.id, + total=result.total, + severity=result.severity_code, + complete=result.complete, + ts=ctx.now, + # PHQ-9 item 9 is stored separately: any value >= 1 escalates regardless of the total. + item9=answers.get("phq9_9") if scale.id == "phq9" else None, + scheduled=bool(session.get("scheduled", True)), + flags={f: True for f in result.flags}, + ) + sessions = {k: v for k, v in (ctx.data.get("scales") or {}).items() if k != scale.id} + done = Step( + replies=[Reply(text=ctx.texts.scale.done)], + next_state=session["return_state"], + data={"scales": sessions}, + notify_monitoring=True, + ) + if not crisis_flags: + return done + + # Crisis path decided by the scale's own flags (deterministic), not by the LLM. + risk_screen = ctx.texts.settings.risk_screen_scale + if crisis_open(ctx.data): + crisis = Step(next_state="crisis") + else: + crisis = open_crisis(ctx, categories=[], source=f"scale:{scale.id}") + step = done.then(crisis) + if scale.id != 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 new file mode 100644 index 0000000..6723cb8 --- /dev/null +++ b/app/dialog/texts.py @@ -0,0 +1,121 @@ +"""Patient-facing dialog texts from `config/texts/dialog.yaml` (ADR-008: texts live in config, not code).""" + +from functools import lru_cache +from pathlib import Path + +import yaml +from pydantic import BaseModel, ConfigDict, Field + +from app.config import get_settings + + +class _Strict(BaseModel): + # A typo in a YAML key must fail at load time, not silently show an empty message to a patient. + model_config = ConfigDict(extra="forbid", frozen=True) + + +class DialogSettings(_Strict): + cssrs_cooldown_hours: float = 24 + max_safety_events: int = 50 + crisis_flags: list[str] = Field(default_factory=list) + risk_screen_scale: str = "cssrs_screen" + + +class Commands(_Strict): + revoke: list[str] + + +class ConsentTexts(_Strict): + version: str + text: str + accept: str + decline: str + refused: str + reconsider: str + code_invalid: str + + +class AgeGateTexts(_Strict): + question: str + adult: str + minor: str + unclear: str + minor_stop: str + + +class IdleButtons(_Strict): + checkin: str + med_question: str + contact_doctor: str + revoke: str + + +class IdleTexts(_Strict): + welcome: str + menu_prompt: str + buttons: IdleButtons + stub_checkin: str + stub_med_question: str + contact_doctor_ack: str + visit_stub: str + + +class RevokedTexts(_Strict): + confirm: str + + +class CrisisTexts(_Strict): + repeat: str + hold: str + closed: str + + +class ScaleTexts(_Strict): + intro: str + question: str + clarify: str + buttons_only: str + done: str + risk_screen_intro: str + + +class SafetyPlanTexts(_Strict): + title: str + template: list[str] + means_restriction: str + + +class ReachOutTexts(_Strict): + silence: str + missed_doses: str + default: str + + +class VisitTexts(_Strict): + button: str + + +class DialogTexts(_Strict): + status: str + settings: DialogSettings + commands: Commands + consent: ConsentTexts + age_gate: AgeGateTexts + idle: IdleTexts + revoked: RevokedTexts + crisis: CrisisTexts + scale: ScaleTexts + safety_plan: SafetyPlanTexts + reach_out: ReachOutTexts + visit: VisitTexts + unavailable: str + + +def dialog_texts_path() -> Path: + return get_settings().config_dir / "texts" / "dialog.yaml" + + +@lru_cache +def load_dialog_texts(path: Path | None = None) -> DialogTexts: + path = path or dialog_texts_path() + return DialogTexts.model_validate(yaml.safe_load(path.read_text(encoding="utf-8"))) diff --git a/config/texts/dialog.yaml b/config/texts/dialog.yaml new file mode 100644 index 0000000..f295afd --- /dev/null +++ b/config/texts/dialog.yaml @@ -0,0 +1,114 @@ +# Тексты диалогового движка (issue #10): согласие, возраст, меню, кризис, шкалы, отзыв согласия. +# СТАТУС: черновик, утверждает психиатр (#8). +# +# Правила текстов: без диагнозов и названий расстройств; без советов по дозам; не обещать конфиденциальность +# сверх реальной; не использовать слова «учёт», «контроль», «психиатрический надзор» (проверяется тестом). +# Кризисный скрипт — в crisis.yaml (отправляется дословно, LLM не вызывается). + +status: "черновик, утверждает психиатр (#8)" + +settings: + # Повторный C-SSRS по сигналу POSSIBLE не чаще, чем раз в N часов (сигнал всё равно пишется в safety_events). + cssrs_cooldown_hours: 24 + # Сколько последних safety-событий хранить в dialog_state.data["safety_events"]. + max_safety_events: 50 + # Флаги шкал (config/scales/*.yaml), при которых сессия шкалы сразу уходит в кризисный путь. + crisis_flags: [phq9_item9_positive, cssrs_high] + # Шкала-скрининг суицидального риска (POSSIBLE и положительный п.9 PHQ-9 → эта шкала). + risk_screen_scale: cssrs_screen + +commands: + # Текстовые команды отзыва согласия (сравниваются после приведения к нижнему регистру). + revoke: ["/revoke", "отозвать согласие", "отзываю согласие"] + +consent: + version: "v1-draft-2026-10-03" + text: |- + Здравствуйте! Я — цифровой помощник, который сопровождает вас между визитами к врачу: напоминаю о приёме препарата, спрашиваю о самочувствии и передаю важное вашему лечащему врачу. + + Прежде чем начать, мне нужно ваше согласие на обработку персональных данных, в том числе сведений о здоровье (152-ФЗ «О персональных данных», ст. 9 и 10). + + • Что обрабатывается: ваши ответы о самочувствии, сне, приёме препарата и результаты коротких опросников. Перед автоматической обработкой из сообщений удаляются имена, телефоны и другие личные данные. + • Кто видит данные: только ваш лечащий врач. Данные хранятся в России и не передаются третьим лицам без вашего согласия, кроме случаев, прямо предусмотренных законом. + • Я не ставлю диагнозов и не заменяю врача. Решения о лечении принимает только врач. + • Участие добровольное. Отказ никак не влияет на вашу медицинскую помощь. + • Отозвать согласие можно в любой момент в один шаг: кнопкой «Отозвать согласие» или командой /revoke. + + Я не служба экстренной помощи. Если вам угрожает опасность прямо сейчас — звоните 112. + accept: "Согласен(на)" + decline: "Не согласен(на)" + refused: |- + Понимаю, это ваше право. Без согласия я не обрабатываю и не сохраняю ваши сообщения. Ваше лечение у врача продолжается как обычно. + Если передумаете — нажмите кнопку ниже или отправьте /start. + reconsider: "Я передумал(а)" + code_invalid: "Код подключения не найден или уже использован. Пожалуйста, уточните код у вашего врача и отправьте его ещё раз." + +age_gate: + question: "Уточните, пожалуйста: вам уже исполнилось 18 лет? Можно нажать кнопку или написать год рождения (например, 1990)." + adult: "Да, мне 18 или больше" + minor: "Нет, мне меньше 18" + unclear: "Не удалось понять ответ. Нажмите кнопку или напишите год рождения четырьмя цифрами, например 1990." + minor_stop: |- + Спасибо, что ответили. Этот помощник рассчитан на взрослых, поэтому дальше я не смогу с вами работать и не сохраняю ваши ответы. + Если вам тяжело или нужна поддержка — позвоните на Детский телефон доверия 8-800-2000-122 (бесплатно, анонимно, круглосуточно) или расскажите взрослому, которому доверяете: родителям, учителю, школьному психологу. + Если есть угроза жизни прямо сейчас — звоните 112. + +idle: + welcome: |- + Спасибо, всё готово! Я буду на связи каждый день: напомню о препарате и спрошу о самочувствии. Если захотите что-то рассказать — просто напишите или выберите кнопку ниже. + menu_prompt: "Чем могу помочь?" + buttons: + checkin: "Как я сегодня" + med_question: "Вопрос о лекарстве" + contact_doctor: "Связаться с врачом" + revoke: "Отозвать согласие" + # Заглушки кнопок меню; ветки ветерана (#36/#37) и med_qa (#15) подменяют их своими триггерами. + stub_checkin: "Короткий опрос о самочувствии скоро появится здесь. А пока можно просто написать, как вы себя чувствуете сегодня." + stub_med_question: "Напишите ваш вопрос о препарате. Я отвечаю только по официальной инструкции, а если ответа в ней нет — передам вопрос врачу." + contact_doctor_ack: "Я отметил вашу просьбу связаться с врачом — она появится в вашей карточке у врача. Если ситуация срочная, звоните 112." + visit_stub: "Онлайн-запись появится здесь позже. Пока вы можете записаться к своему врачу через регистратуру клиники." + +revoked: + confirm: |- + Согласие отозвано. Я больше не буду писать вам и обрабатывать ваши сообщения, а ваши данные помечены к удалению. + Если передумаете — отправьте /start. Если вам угрожает опасность — звоните 112. + +crisis: + # Повторный острый сигнал в уже открытом кризисе: скрипт второй раз не шлём, только это + ресурсы. + repeat: "Я рядом. Ваш врач уже уведомлён. Если вам угрожает опасность прямо сейчас — позвоните:" + # Любое другое сообщение, пока кризис открыт (до подтверждения врача). + hold: "Я на связи и читаю вас. Ваш врач в курсе и свяжется с вами. Если станет хуже — звоните 112 или на телефон доверия 8-800-2000-122." + # Врач подтвердил кризисный алерт (doctor_ack): выходим из кризиса. + closed: "Ваш врач получил сообщение и в курсе вашей ситуации. Спасибо, что остаётесь на связи. Если снова станет тяжело — напишите мне в любой момент." + +scale: + intro: "Несколько коротких вопросов ({count}). {instruction}" + question: "{n}/{total}. {text}" + clarify: "Не совсем понял ответ. Выберите, пожалуйста, вариант кнопкой или опишите, как часто это было." + buttons_only: "На этот вопрос, пожалуйста, ответьте кнопкой." + done: "Спасибо, ответы сохранены. Ваш врач увидит результаты." + risk_screen_intro: "Спасибо, что поделились. Мне важно задать несколько коротких вопросов, чтобы лучше понять, как вы сейчас." + +safety_plan: + title: "Ваш план безопасности:" + # Шаблон Стэнли–Брауна, если план ещё не составлен с врачом на визите. + template: + - "Мои тревожные сигналы: что я замечаю, когда становится хуже." + - "Что помогает мне самому: прогулка, душ, дыхание, любимое дело." + - "Люди и места, которые отвлекают: к кому можно пойти." + - "Кому позвонить, если тяжело: близкий человек, сослуживец." + - "Специалисты: ваш врач, телефон доверия 8-800-2000-122, экстренные службы 112." + - "Безопасная среда: оружие и лишние лекарства — на хранение близкому человеку." + means_restriction: "Если дома есть оружие или запас лекарств — попросите близкого человека временно хранить их у себя." + +reach_out: + silence: "Давно не было вестей от вас. Как вы? Можно ответить одним словом — мне важно знать, что с вами всё в порядке." + missed_doses: "Заметил, что несколько приёмов препарата пропущено. Всё ли в порядке? Если что-то мешает — расскажите, я передам врачу." + default: "Хотел узнать, как вы. Напишите пару слов, когда будет удобно." + +visit: + button: "Записаться к врачу раньше" + +unavailable: |- + Сейчас у меня технические неполадки, и я не могу сохранить ваше сообщение. Пожалуйста, напишите чуть позже. + Если вам плохо прямо сейчас — звоните 112 или на телефон доверия 8-800-2000-122. diff --git a/tests/test_dialog_engine.py b/tests/test_dialog_engine.py new file mode 100644 index 0000000..8063bdc --- /dev/null +++ b/tests/test_dialog_engine.py @@ -0,0 +1,485 @@ +"""Dialog engine (#10): consent, age gate, crisis without LLM, C-SSRS on POSSIBLE, scale sessions, revocation. + +Two in-memory SQLite databases (identity / clinical), offline mock LLM wrapped in a spy that records every task +call, and a fake monitoring service (lane B stub) that records calls and returns scripted actions. +""" + +from collections.abc import AsyncIterator +from typing import Any + +import pytest +from sqlalchemy import select + +from app.config import get_settings +from app.core import clock +from app.core.rules import Action +from app.core.safety import crisis_reply +from app.core.types import AlertLevel, ScrubbedText +from app.db import identity_repo, repo, session +from app.db.bootstrap import init_db +from app.db.clinical import ItemResponse, Message, SafetyPlan +from app.dialog.contracts import Event, Reply, get_engine +from app.dialog.engine import Engine +from app.dialog.fsm import Ctx, Step, on_enter +from app.dialog.texts import load_dialog_texts +from app.integrations.base import VerifiedPerson, integrations +from app.llm.gateway import LLMGateway +from app.llm.mock import MockTasks +from app.llm.schemas import RiskVerdict + +TOKEN = "a" * 32 +TEXTS = load_dialog_texts() +MEM = "sqlite+aiosqlite:///:memory:" + + +class SpyTasks(MockTasks): + """Offline mock that records which LLM tasks were called (to prove the crisis path never calls the LLM).""" + + def __init__(self, risk: str | None = None) -> None: + self.calls: list[str] = [] + self._risk = risk + + async def extract_item(self, scale_item_id: str, item_text: str, max_score: int, answer: ScrubbedText) -> Any: + self.calls.append("extract_item") + return await super().extract_item(scale_item_id, item_text, max_score, answer) + + async def extract_checkin(self, text: ScrubbedText) -> Any: + self.calls.append("extract_checkin") + return await super().extract_checkin(text) + + async def risk_classify(self, text: ScrubbedText) -> RiskVerdict | None: + self.calls.append("risk_classify") + if self._risk is not None: + return RiskVerdict(level=self._risk, confidence=0.9) # type: ignore[arg-type] + return await super().risk_classify(text) + + async def phrase(self, intent: str, payload: ScrubbedText, base_text: str) -> str | None: + self.calls.append("phrase") + return base_text + + +class FakeMonitoring: + def __init__(self, actions: list[Action] | None = None, in_crisis: bool = True) -> None: + self.calls = 0 + self.actions = actions or [] + self.in_crisis = in_crisis + + async def after_data_change(self, token: str) -> list[Action]: + self.calls += 1 + out, self.actions = self.actions, [] + return out + + async def is_in_crisis(self, token: str) -> bool: + return self.in_crisis + + +@pytest.fixture +async def dbs() -> AsyncIterator[None]: + await session.configure(identity_url=MEM, clinical_url=MEM) + await init_db() + async with session.get_identity_session() as s: + await identity_repo.create_patient( + s, + token=TOKEN, + display_name="Тестовый Пациент", + snils="000-000-000 00", + clinic_id="c1", + doctor_id="doc-test", + now=clock.now(), + birth_year=1992, + sex="m", + ) + async with session.get_clinical_session() as s: + await repo.upsert_profile(s, token=TOKEN, doctor_id="doc-test", modules=[], sex="m") + yield + await session.configure() # back to default URLs for other test modules + + +@pytest.fixture +def spy() -> SpyTasks: + return SpyTasks() + + +@pytest.fixture +def mon() -> FakeMonitoring: + return FakeMonitoring() + + +@pytest.fixture +def engine(spy: SpyTasks, mon: FakeMonitoring) -> Engine: + return Engine(gateway=LLMGateway(spy), monitoring=mon) + + +# ----------------------------------------------------------------------------------------------- helpers + + +async def text(e: Engine, t: str) -> list[Reply]: + return await e.handle(TOKEN, Event(type="text", text=t)) + + +async def press(e: Engine, data: str) -> list[Reply]: + return await e.handle(TOKEN, Event(type="button", button=data)) + + +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 set_state(state: str, **data: Any) -> None: + async with session.get_clinical_session() as s: + await repo.set_dialog_state(s, token=TOKEN, state=state, data={"consent": True, **data}, now=clock.now()) + + +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 stored_messages() -> list[Message]: + async with session.get_clinical_session() as s: + return list((await s.scalars(select(Message).where(Message.token == TOKEN).order_by(Message.id))).all()) + + +async def identity_patient() -> Any: + async with session.get_identity_session() as s: + return await identity_repo.get_patient(s, TOKEN) + + +async def set_birth_year(year: int | None) -> None: + async with session.get_identity_session() as s: + patient = await identity_repo.get_patient(s, TOKEN) + assert patient is not None + patient.birth_year = year + + +def buttons(reply: Reply) -> list[str]: + return [b.data for row in reply.buttons for b in row] + + +# A fake module onboarding entry point, to check the hand-over without depending on #36. +@on_enter("zztest.onboarding.start") +async def _zz_onboarding(ctx: Ctx) -> Step: + return Step(replies=[Reply(text="zz onboarding")]) + + +# ----------------------------------------------------------------------------------------------- registration + + +async def test_get_engine_returns_real_engine() -> None: + assert isinstance(get_engine(), Engine) + + +async def test_consent_accept_goes_to_idle(dbs: None, engine: Engine) -> None: + r = await engine.handle(TOKEN, Event(type="start")) + assert "152-ФЗ" in r[0].text and buttons(r[0]) == ["consent:yes", "consent:no"] + assert (await get_state())[0] == "consent" + + r = await press(engine, "consent:yes") + state, data = await get_state() + # Birth year 1992 is known from registration → adult without asking; no module flow → idle. + assert state == "idle" and data["consent"] is True + assert r[-1].text.startswith(TEXTS.idle.welcome[:20]) and "consent:revoke" in buttons(r[-1]) + patient = await identity_patient() + assert patient.consent_at is not None and patient.consent_ver == TEXTS.consent.version + + +async def test_connect_code_verified(dbs: None, engine: Engine, monkeypatch: pytest.MonkeyPatch) -> None: + class Ident: + async def verify(self, auth_code: str) -> VerifiedPerson | None: + return VerifiedPerson(token=TOKEN, doctor_id="doc-test") if auth_code == "GOOD" else None + + monkeypatch.setattr(integrations, "identity", Ident()) + r = await engine.handle(TOKEN, Event(type="start", params={"code": "BAD"})) + assert r[0].text == TEXTS.consent.code_invalid and (await get_state())[0] == "new" + r = await engine.handle(TOKEN, Event(type="start", params={"code": "GOOD"})) + assert buttons(r[0]) == ["consent:yes", "consent:no"] + + +async def test_consent_refused_no_processing(dbs: None, engine: Engine, spy: SpyTasks) -> None: + await engine.handle(TOKEN, Event(type="start")) + r = await press(engine, "consent:no") + assert r[0].text == TEXTS.consent.refused and (await get_state())[0] == "refused" + r = await text(engine, "мне плохо, не сплю, позвоните 89161234567") + assert r[0].text == TEXTS.consent.refused + assert spy.calls == [] # nothing reached the LLM + assert await stored_messages() == [] # nothing stored without consent + assert (await identity_patient()).consent_at is None + r = await press(engine, "consent:restart") + assert buttons(r[0]) == ["consent:yes", "consent:no"] + + +async def test_minor_gets_child_helpline_and_stop(dbs: None, engine: Engine, spy: SpyTasks) -> None: + await set_birth_year(None) + await engine.handle(TOKEN, Event(type="start")) + r = await press(engine, "consent:yes") + assert buttons(r[-1]) == ["age:adult", "age:minor"] + r = await text(engine, "2012") + assert "8-800-2000-122" in r[0].text + state, data = await get_state() + assert state == "minor" and data["consent"] is False + assert (await identity_patient()).revoked_at is not None # a minor's consent is withdrawn + calls = len(spy.calls) + r = await text(engine, "а что дальше?") + assert "8-800-2000-122" in r[0].text and len(spy.calls) == calls + + +async def test_age_gate_button_adult(dbs: None, engine: Engine) -> None: + await set_birth_year(None) + await engine.handle(TOKEN, Event(type="start")) + await press(engine, "consent:yes") + r = await text(engine, "не скажу") + assert r[0].text.startswith(TEXTS.age_gate.unclear) + await press(engine, "age:adult") + assert (await get_state())[0] == "idle" + + +async def test_hand_over_to_module_onboarding(dbs: None, engine: Engine) -> None: + async with session.get_clinical_session() as s: + await repo.upsert_profile(s, token=TOKEN, modules=["zztest"]) + await engine.handle(TOKEN, Event(type="start")) + r = await press(engine, "consent:yes") + assert r[-1].text == "zz onboarding" + assert (await get_state())[0] == "zztest.onboarding.start" + + +# ----------------------------------------------------------------------------------------------- safety + + +async def test_acute_sends_script_without_any_llm_call( + dbs: None, engine: Engine, spy: SpyTasks, mon: FakeMonitoring +) -> None: + await set_state("idle") + r = await text(engine, "хочу покончить с собой") + assert len(r) == 1 and r[0].kind == "crisis" + assert r[0].text == crisis_reply(["suicide"]).text # verbatim fixed script + assert spy.calls == [] # lexicon ACUTE → risk_classify is not even called + state, data = await get_state() + assert state == "crisis" and data["crisis"]["open"] is True + event = data["safety_events"][-1] + assert event["level"] == "acute" and event["categories"] == ["suicide"] and event["handled"] == "crisis_script" + assert mon.calls == 1 # lane B gets the chance to escalate + + +async def test_repeat_crisis_does_not_duplicate_script(dbs: None, engine: Engine, spy: SpyTasks) -> None: + await set_state("idle") + script = (await text(engine, "хочу покончить с собой"))[0].text + r = await text(engine, "не хочу жить") + assert r[0].kind == "crisis" and r[0].text != script + assert r[0].text.startswith(TEXTS.crisis.repeat) and "112" in r[0].text + assert spy.calls == [] + r = await text(engine, "ладно") + assert r[0].text == TEXTS.crisis.hold + _, data = await get_state() + assert [e["handled"] for e in data["safety_events"]] == ["crisis_script", "crisis_repeat"] + + +async def test_doctor_ack_leaves_crisis(dbs: None, engine: Engine) -> None: + await set_state("idle") + await text(engine, "хочу покончить с собой") + r = await system(engine, "doctor_ack") + assert r[0].text == TEXTS.crisis.closed + state, data = await get_state() + assert state == "idle" and data["crisis"]["open"] is False + # A new acute message after the crisis was closed gets the full script again. + r = await text(engine, "хочу покончить с собой") + assert r[0].text == crisis_reply(["suicide"]).text + + +async def test_crisis_left_when_monitoring_says_resolved(dbs: None, spy: SpyTasks) -> None: + mon = FakeMonitoring(in_crisis=False) + e = Engine(gateway=LLMGateway(spy), monitoring=mon) + await set_state("crisis", crisis={"open": True}) + r = await text(e, "мне получше") + assert r[0].text == TEXTS.crisis.closed and (await get_state())[0] == "idle" + + +async def test_possible_starts_cssrs_and_returns(dbs: None, engine: Engine, spy: SpyTasks) -> None: + await set_state("idle") + r = await text(engine, "всё бессмысленно, устал от всего") + assert "risk_classify" in spy.calls + assert (await get_state())[0] == "scale.cssrs_screen" + assert r[-1].kind == "question" and [b.label for row in r[-1].buttons for b in row] == ["Нет", "Да"] + assert TEXTS.scale.risk_screen_intro in r[-1].text + + r = await text(engine, "нет") # C-SSRS is buttons-only: free text is never sent to extract_item + assert r[0].text.startswith(TEXTS.scale.buttons_only) and "extract_item" not in spy.calls + + for i in range(1, 7): + r = await press(engine, f"scale:cssrs_screen:cssrs_{i}:0") + state, data = await get_state() + assert state == "idle" and "cssrs_screen" not in data.get("scales", {}) + assert r[0].text == TEXTS.scale.done + async with session.get_clinical_session() as s: + (res,) = await repo.list_scale_results(s, TOKEN, "cssrs_screen") + assert res.severity == "none" and res.scheduled is False and res.complete + assert data["safety_events"][-1]["handled"] == "risk_screen" + + +async def test_cssrs_high_goes_to_crisis(dbs: None, engine: Engine) -> None: + await set_state("idle") + await text(engine, "всё бессмысленно, устал от всего") + await press(engine, "scale:cssrs_screen:cssrs_1:1") + await press(engine, "scale:cssrs_screen:cssrs_2:1") + await press(engine, "scale:cssrs_screen:cssrs_3:0") + r = await press(engine, "scale:cssrs_screen:cssrs_4:1") # intent → high: the session ends at once + assert any(x.kind == "crisis" and x.text == crisis_reply([]).text for x in r) + assert (await get_state())[0] == "crisis" + + +async def test_llm_signal_can_raise_risk(dbs: None, mon: FakeMonitoring) -> None: + spy = SpyTasks(risk="possible") + e = Engine(gateway=LLMGateway(spy), monitoring=mon) + await set_state("idle") + await text(e, "обычный день") # lexicon NONE, LLM says possible → max() = POSSIBLE + state, data = await get_state() + assert state == "scale.cssrs_screen" and data["safety_events"][-1]["source"] == "llm" + + +# ----------------------------------------------------------------------------------------------- scale session + + +async def test_scale_session_buttons_free_text_and_clarify(dbs: None, engine: Engine, spy: SpyTasks) -> None: + await set_state("idle") + r = await system(engine, "scale_due", scale="phq9", scheduled=True) + assert (await get_state())[0] == "scale.phq9" + assert r[0].kind == "question" and len(buttons(r[0])) == 4 and "1/9" in r[0].text + + await press(engine, "scale:phq9:phq9_1:1") + r = await text(engine, "почти каждый день") # mock extract: confident score 3 + assert "3/9" in r[0].text + r = await text(engine, "ну такое") # no cue → low confidence → clarify, nothing stored + assert r[0].text.startswith(TEXTS.scale.clarify) and "3/9" in r[0].text + r = await press(engine, "scale:phq9:phq9_1:2") # stale button from an earlier question → re-ask + assert "3/9" in r[0].text + for i in range(3, 9): + await press(engine, f"scale:phq9:phq9_{i}:0") + r = await text(engine, "нет") # item 9 is buttons-only + assert r[0].text.startswith(TEXTS.scale.buttons_only) + assert spy.calls.count("extract_item") == 2 + r = await press(engine, "scale:phq9:phq9_9:0") + assert r[0].text == TEXTS.scale.done and (await get_state())[0] == "idle" + + async with session.get_clinical_session() as s: + (res,) = await repo.list_scale_results(s, TOKEN, "phq9") + items = (await s.scalars(select(ItemResponse).order_by(ItemResponse.id))).all() + assert (res.total, res.item9, res.complete, res.scheduled) == (4, 0, True, True) + assert [(i.item_id, i.source) for i in items[:2]] == [("phq9_1", "button"), ("phq9_2", "llm")] + assert items[1].score == 3 and items[1].confidence is not None + + +async def test_phq9_item9_positive_crisis_then_cssrs(dbs: None, engine: Engine, spy: SpyTasks) -> None: + await set_state("idle") + await system(engine, "scale_due", scale="phq9", scheduled=False) + for i in range(1, 9): + await press(engine, f"scale:phq9:phq9_{i}:0") + r = await press(engine, "scale:phq9:phq9_9:2") + kinds = [x.kind for x in r] + assert "crisis" in kinds and r[kinds.index("crisis")].text == crisis_reply([]).text + assert r[-1].kind == "question" and "cssrs" not in r[-1].text # C-SSRS q1 asked after the script + state, data = await get_state() + assert state == "scale.cssrs_screen" and data["crisis"]["open"] is True + assert data["scales"]["cssrs_screen"]["return_state"] == "crisis" + async with session.get_clinical_session() as s: + (res,) = await repo.list_scale_results(s, TOKEN, "phq9") + assert res.item9 == 2 and res.scheduled is False and res.flags == {"phq9_item9_positive": True} + assert "extract_item" not in spy.calls + # A repeated acute phrase during the C-SSRS: short text, no second script, the screen continues. + r = await text(engine, "не хочу жить") + assert r[0].text.startswith(TEXTS.crisis.repeat) and (await get_state())[0] == "scale.cssrs_screen" + + +# ----------------------------------------------------------------------------------------------- monitoring actions + + +async def test_monitoring_actions_executed(dbs: None, spy: SpyTasks) -> None: + mon = FakeMonitoring( + actions=[ + Action(type="alert", level=AlertLevel.HIGH, code="x"), + Action(type="show_safety_plan", code="phq9_item9"), + Action(type="schedule_scale", code="mood_low_3d", params={"scale": "phq9", "scheduled": False}), + ] + ) + e = Engine(gateway=LLMGateway(spy), monitoring=mon) + async with session.get_clinical_session() as s: + s.add(SafetyPlan(token=TOKEN, steps=["Позвонить брату Олегу", "Прогулка"], updated_at=clock.now())) + await set_state("idle") + r = await press(e, "menu:contact_doctor") # records a contact request → monitoring runs + assert r[0].text == TEXTS.idle.contact_doctor_ack + plan = next(x for x in r if x.kind == "safety_plan") + assert "Позвонить брату Олегу" in plan.text + assert r[-1].kind == "question" # unscheduled PHQ-9 started right away (idle = resting) + state, data = await get_state() + assert state == "scale.phq9" and data["scales"]["phq9"]["scheduled"] is False + assert data["contact_requests"] + # Stored outbound texts are scrubbed (the plan names a relative). + assert not any("Олег" in m.text_scrubbed for m in await stored_messages()) + + +async def test_suggest_visit_and_med_info_phrased(dbs: None, spy: SpyTasks) -> None: + mon = FakeMonitoring( + actions=[ + Action(type="med_info", code="alcohol_3d", params={"topic": "alcohol_interaction"}), + Action(type="suggest_visit", code="alcohol_3d", params={"reason": "alcohol"}), + ] + ) + e = Engine(gateway=LLMGateway(spy), monitoring=mon) + await set_state("idle") + r = await press(e, "menu:contact_doctor") + assert any("алкогол" in x.text for x in r) and buttons(r[-1]) == ["menu:visit"] + + +# ------------------------------------------------------------------------------------ consent revocation, privacy + + +async def test_revoke_consent_then_silence(dbs: None, engine: Engine, spy: SpyTasks) -> None: + await set_state("idle") + r = await press(engine, "consent:revoke") + assert r[0].text == TEXTS.revoked.confirm + state, data = await get_state() + assert state == "consent_revoked" and data["consent"] is False and data["delete_requested"] is True + assert (await identity_patient()).revoked_at is not None + before = len(await stored_messages()) + assert await text(engine, "привет, как дела?") == [] + assert await system(engine, "daily_checkin") == [] + assert spy.calls == [] and len(await stored_messages()) == before + r = await engine.handle(TOKEN, Event(type="start")) # coming back = asking for consent again + assert buttons(r[0]) == ["consent:yes", "consent:no"] + + +async def test_revoke_by_text_command(dbs: None, engine: Engine) -> None: + await set_state("idle") + r = await text(engine, "Отозвать согласие") + assert r[0].text == TEXTS.revoked.confirm and (await get_state())[0] == "consent_revoked" + + +async def test_messages_stored_scrubbed(dbs: None, engine: Engine) -> None: + await set_state("idle") + r = await text(engine, "Перезвоните мне на 8 916 123-45-67, сегодня обычный день") + assert r and buttons(r[0]) # neutral support reply with the idle menu + msgs = await stored_messages() + inbound = [m.text_scrubbed for m in msgs if m.direction == "in"] + assert inbound and "[PHONE]" in inbound[0] + assert not any("123-45-67" in m.text_scrubbed or "916" in m.text_scrubbed for m in msgs) + assert any(m.direction == "out" for m in msgs) + + +async def test_db_unavailable_still_answers_crisis(engine: Engine, tmp_path: Any) -> None: + await session.configure(identity_url=MEM, clinical_url=f"sqlite+aiosqlite:///{tmp_path}/missing/dir/x.db") + try: + r = await text(engine, "хочу покончить с собой") + assert r[0].kind == "crisis" and "112" in r[0].text + assert crisis_reply(["suicide"]).doctor_notified not in r[0].text # nobody could be notified + r = await text(engine, "привет") + assert r[0].text == TEXTS.unavailable + finally: + await session.configure() + + +def test_dialog_texts_avoid_forbidden_words() -> None: + raw = (get_settings().config_dir / "texts" / "dialog.yaml").read_text(encoding="utf-8").lower() + body = "\n".join(line for line in raw.splitlines() if not line.lstrip().startswith("#")) + for word in ("учёт", "учет", "контрол", "надзор", "диагноз:"): + assert word not in body + assert "152-фз" in body and "8-800-2000-122" in body -- 2.54.0