"""Daily check-in and scheduled-scale intros (#37, ADR-002, CLINICAL_MODULES §1–2, case «мониторинг после обращения»). Check-in = core (mood 0–10, dose taken) + the module question (veteran: sleep hours + nightmares; alcohol for veterans and for anyone with a positive AUDIT-C) + free text: system:daily_checkin / menu «Как я сегодня» ──▶ daily.mood buttons 0–10 (a bare number is accepted too) ──▶ daily.med only with a prescription and no dose answer today (the reminder may have got it) ──▶ daily.sleep <4 / 4–6 / 6–8 / >8 h ┐ module veteran ──▶ daily.nightmares yes / no ┘ ──▶ daily.alcohol none / 1–2 / more (veteran or AUDIT-C positive) ──▶ daily.text «Как вы? Пару слов» (skip button) → extract_checkin on the SCRUBBED text ──▶ [daily.alcohol_amount] alcohol mentioned in the text but not asked by button → amount buttons ──▶ idle upsert done, `notify_monitoring` → rules (lane B) → dialog actions Data algorithm (medical data, jury requirement): * Every button answer is written to `daily_checkins` immediately (`repo.upsert_checkin` merges non-None fields), so an interrupted check-in (crisis, the patient walking away) still leaves today's mood on record — the "mood ≤ 3 three days in a row" rule must not miss a day because the free text was never written. * Button answers are the patient's own and always win over the LLM: from the free text the LLM may only add side effects, the quit intent and fields that were not asked by buttons today. * The LLM never decides anything here: alerts and follow-ups (delayed-onset explanation, unscheduled PHQ-9, alcohol interaction, visit offer) come from the rules through `MonitoringService` (ADR-007). * A free-text message classified POSSIBLE by the safety layer is taken over by the engine (C-SSRS). When the screen returns to `daily.text` the message counts as the answer: the check-in is closed, not asked again. * A check-in left unfinished is dropped on the next event of a later day, or after `settings.abandon_after_hours` of silence (engine abandon timeout, #56) — answers given so far stay stored. Also here: friendly intros for scheduled scales (pcl5, phq9, dar5, auditc), registered with `register_scale_intro` so the engine's `scale_due` / rule-triggered sessions show them. """ import re from datetime import datetime from functools import lru_cache from pathlib import Path from typing import Any import yaml from pydantic import BaseModel, ConfigDict from app.config import get_settings from app.core.intake_runtime import intake_enabled from app.core.rules import load_rules from app.db import repo from app.dialog.contracts import Button, Event, Reply from app.dialog.flows import intake from app.dialog.flows.core import menu_reply from app.dialog.flows.knowledge import knowledge_answer, looks_like_question from app.dialog.flows.meds import ( answer_question, checkin_of, current_prescription, leave_to_idle, load_meds_texts, record_dose, record_side_effects, run_trigger_inline, save_checkin, side_effect_reply, ) from app.dialog.fsm import Ctx, Step, abandonable, on_enter, state, trigger from app.dialog.scale_session import register_scale_intro PREFIX = "daily" STEPS = ("mood", "med", "sleep", "nightmares", "alcohol", "text") ALCOHOL_AMOUNT = "alcohol_amount" SLEEP_BUCKETS = ("<4", "4-6", "6-8", ">8") ALCOHOL_BUCKETS = ("none", "1-2", "more") _NUMBER = re.compile(r"^\s*(\d{1,2})\s*$") # ------------------------------------------------------------------------------------------- texts class _Strict(BaseModel): model_config = ConfigDict(extra="forbid", frozen=True) class DailySettings(_Strict): sleep_modules: list[str] alcohol_daily_modules: list[str] intro_scales: list[str] class Question(_Strict): question: str class OptionsQuestion(_Strict): question: str options: dict[str, str] class YesNoQuestion(_Strict): question: str yes: str no: str class AlcoholQuestion(_Strict): question: str clarify: str options: dict[str, str] class TextQuestion(_Strict): question: str skip: str class CheckinTexts(_Strict): start: str mood: Question sleep: OptionsQuestion nightmares: YesNoQuestion alcohol: AlcoholQuestion text: TextQuestion buttons_hint: str done: str already_done: str class VisitTexts(_Strict): requested: str class ScaleIntros(_Strict): scheduled: dict[str, str] unscheduled: dict[str, str] class DailyTexts(_Strict): status: str settings: DailySettings checkin: CheckinTexts visit: VisitTexts scale_intros: ScaleIntros def daily_texts_path() -> Path: return get_settings().config_dir / "texts" / "veteran_daily.yaml" @lru_cache def load_daily_texts(path: Path | None = None) -> DailyTexts: path = path or daily_texts_path() return DailyTexts.model_validate(yaml.safe_load(path.read_text(encoding="utf-8"))) # ------------------------------------------------------------------------------------------- step logic def _today(ctx: Ctx) -> str: return ctx.now.date().isoformat() def _ck(ctx: Ctx) -> dict[str, Any]: return dict(ctx.data.get("checkin") or {}) async def _modules(ctx: Ctx) -> set[str]: profile = await ctx.profile() return set(profile.modules or []) if profile is not None else set() async def _auditc_positive(ctx: Ctx) -> bool: """Latest AUDIT-C at or above the sex-specific cut-off of rule `auditc_positive` (one threshold, one place).""" results = await repo.list_scale_results(ctx.session, ctx.token, "auditc") if not results: return False rule = next((r for r in load_rules().rules if r.id == "auditc_positive"), None) if rule is None: return False profile = await ctx.profile() key = "threshold_f" if profile is not None and profile.sex == "f" else "threshold_m" return results[-1].total >= int(rule.params[key]) async def _applicable(ctx: Ctx, step: str) -> bool: settings = load_daily_texts().settings if step in ("mood", "text"): return True if step == "med": today = await checkin_of(ctx, ctx.now.date()) return (today is None or today.med_taken is None) and await current_prescription(ctx) is not None modules = await _modules(ctx) if step in ("sleep", "nightmares"): return bool(modules & set(settings.sleep_modules)) if step == "alcohol": return bool(modules & set(settings.alcohol_daily_modules)) or await _auditc_positive(ctx) return False async def _next_step(ctx: Ctx, after: str | None) -> str: rest = STEPS if after is None else STEPS[STEPS.index(after) + 1 :] for step in rest: if await _applicable(ctx, step): return step return "text" def _buttons(step: str, options: list[tuple[str, str]]) -> list[list[Button]]: return [[Button(label=label, data=f"{PREFIX}:{step}:{value}")] for value, label in options] async def _question(ctx: Ctx, step: str, note: str = "") -> Reply: t = load_daily_texts().checkin buttons: list[list[Button]] if step == "mood": text = t.mood.question # Two rows (0–5, 6–10): 11 buttons in one column are hard to use on a phone. buttons = [ [Button(label=str(v), data=f"{PREFIX}:mood:{v}") for v in range(6)], [Button(label=str(v), data=f"{PREFIX}:mood:{v}") for v in range(6, 11)], ] elif step == "med": m = load_meds_texts().checkin rx = await current_prescription(ctx) text = m.question.format(drug=rx.drug) if rx is not None else m.question_no_drug buttons = _buttons(step, [("yes", m.yes), ("no", m.no), ("later", m.later)]) elif step == "sleep": text = t.sleep.question buttons = _buttons(step, [(b, t.sleep.options[b]) for b in SLEEP_BUCKETS]) elif step == "nightmares": text = t.nightmares.question buttons = _buttons(step, [("yes", t.nightmares.yes), ("no", t.nightmares.no)]) elif step in ("alcohol", ALCOHOL_AMOUNT): text = t.alcohol.question if step == "alcohol" else t.alcohol.clarify buttons = _buttons(step, [(b, t.alcohol.options[b]) for b in ALCOHOL_BUCKETS]) else: text = t.text.question buttons = _buttons("text", [("skip", t.text.skip)]) return Reply(text="\n\n".join(p for p in (note, text) if p), buttons=buttons, kind="question") async def start_checkin(ctx: Ctx, *, from_menu: bool) -> Step: if ctx.data.get("checkin_done") == _today(ctx): # The scheduler may fire twice; the menu button gets a friendly note instead of a second check-in. return Step(replies=[menu_reply(ctx, load_daily_texts().checkin.already_done)]) if from_menu else Step() first = await _next_step(ctx, None) ck = {"day": _today(ctx), "answered": [], "started_at": ctx.now.isoformat()} return Step( replies=[Reply(text=load_daily_texts().checkin.start)], next_state=f"{PREFIX}.{first}", data={"checkin": ck}, ) def _finish(ctx: Ctx, replies: list[Reply] | None = None) -> Step: done = Reply(text=load_daily_texts().checkin.done) return Step( replies=[done, *(replies or [])], next_state="idle", data={"checkin": None, "checkin_done": _today(ctx)}, notify_monitoring=True, ) def _consumed_by_safety(ctx: Ctx, ck: dict[str, Any]) -> bool: """The free-text answer was taken over by the safety layer (C-SSRS screen) after the question was asked.""" asked = ck.get("text_asked_at") if not asked: return False for event in reversed(ctx.data.get("safety_events") or []): if event.get("state") == f"{PREFIX}.text" and datetime.fromisoformat(event["ts"]) >= datetime.fromisoformat( asked ): return True return False @on_enter(PREFIX) async def _enter(ctx: Ctx) -> Step: step = ctx.state.split(".", 1)[1] if "." in ctx.state else "" ck = _ck(ctx) if not ck or ck.get("day") != _today(ctx): return Step(next_state="idle", data={"checkin": None}) if step == "text": if _consumed_by_safety(ctx, ck): return _finish(ctx) return Step( replies=[await _question(ctx, step)], data={"checkin": {**ck, "text_asked_at": ctx.now.isoformat()}} ) return Step(replies=[await _question(ctx, step)]) async def _advance(ctx: Ctx, step: str, ck: dict[str, Any]) -> Step: ck = {**ck, "answered": [*ck.get("answered", []), step]} if step == ALCOHOL_AMOUNT: return _finish(ctx) nxt = await _next_step(ctx, step) return Step(next_state=f"{PREFIX}.{nxt}", data={"checkin": ck}) async def _record_button(ctx: Ctx, step: str, value: str) -> bool: """Store one button answer right away. False = value not valid for this step (stale/forged button).""" if step == "mood" and value.isdigit() and 0 <= int(value) <= 10: await save_checkin(ctx, mood=int(value)) elif step == "med" and value in ("yes", "no", "later"): if value != "later": # «Ещё нет» is not a missed dose: nothing is recorded yet await record_dose(ctx, value == "yes", status=None if value == "yes" else "missed") elif step == "sleep" and value in SLEEP_BUCKETS: await save_checkin(ctx, sleep_bucket=value) elif step == "nightmares" and value in ("yes", "no"): await save_checkin(ctx, nightmares=value == "yes") elif step in ("alcohol", ALCOHOL_AMOUNT) and value in ALCOHOL_BUCKETS: await save_checkin(ctx, alcohol_bucket=value) else: return False return True async def _free_text(ctx: Ctx, ck: dict[str, Any]) -> Step: assert ctx.text is not None facts = await ctx.gateway.extract_checkin(ctx.text) answered = set(ck.get("answered", [])) fields: dict[str, Any] = {"free_text_scrubbed": ctx.text} replies: list[Reply] = [] clarify_alcohol = False if facts is not None: if facts.quit_intent: fields["quit_intent"] = True # Buttons win: the LLM only fills what was not asked today. if "med" not in answered and facts.med_taken is not None: today = await checkin_of(ctx, ctx.now.date()) if today is None or today.med_taken is None: fields["med_taken"] = facts.med_taken if "sleep" not in answered and facts.sleep_hours_bucket is not None: fields["sleep_bucket"] = facts.sleep_hours_bucket if "nightmares" not in answered and facts.nightmares is not None: fields["nightmares"] = facts.nightmares if "alcohol" not in answered and facts.alcohol is not None: fields["alcohol_bucket"] = facts.alcohol # provisional; the amount buttons below refine it clarify_alcohol = facts.alcohol != "none" if facts.side_effects: await record_side_effects(ctx, [str(c) for c in facts.side_effects]) # One topic per message: when the text also carries a quit intent or alcohol, the rules answer # those (delayed onset / interaction); the side-effect excerpt would only bury that message. if not facts.quit_intent and facts.alcohol in (None, "none"): replies.append(await side_effect_reply(ctx, [str(c) for c in facts.side_effects])) await save_checkin(ctx, **fields) if clarify_alcohol: return Step(replies=replies, next_state=f"{PREFIX}.{ALCOHOL_AMOUNT}", data={"checkin": ck}) return _finish(ctx, replies) abandonable(PREFIX) @state(PREFIX) async def _handle(ctx: Ctx, event: Event) -> Step: step = ctx.state.split(".", 1)[1] if "." in ctx.state else "" ck = _ck(ctx) if not ck or ck.get("day") != _today(ctx): # Unfinished check-in from an earlier day: drop it and handle the event as idle would. return await leave_to_idle(ctx, event, {"checkin": None}) if event.type == "button" and event.button: parts = event.button.split(":") if len(parts) == 3 and parts[0] == PREFIX and parts[1] == step: if step == "text" and parts[2] == "skip": return _finish(ctx) if await _record_button(ctx, step, parts[2]): return await _advance(ctx, step, ck) if parts[0] == "med": # A reminder button pressed mid check-in: answer it, then repeat the current question. inline = await run_trigger_inline(ctx, f"button:{event.button}", event) if inline is not None: return inline if inline.next_state else inline.then(Step(replies=[await _question(ctx, step)])) if parts[0] in (PREFIX, "scale"): return Step(replies=[await _question(ctx, step)]) # stale button of an earlier question return await leave_to_idle(ctx, event, {"checkin": None}) if event.type == "system" and event.system: if event.system == "daily_checkin": return Step(replies=[await _question(ctx, step)]) inline = await run_trigger_inline(ctx, f"system:{event.system}", event) if inline is None: return Step() return inline if inline.next_state else inline.then(Step(replies=[await _question(ctx, step)])) if event.type == "text" and ctx.text is not None: if step == "text": return await _free_text(ctx, ck) number = _NUMBER.match(ctx.text) if step == "mood" and number and await _record_button(ctx, "mood", number.group(1)): return await _advance(ctx, step, ck) return Step(replies=[await _question(ctx, step, note=load_daily_texts().checkin.buttons_hint)]) return Step(replies=[await _question(ctx, step)]) # ------------------------------------------------------------------------------------------- idle triggers @trigger("system:daily_checkin", replace=True) async def _daily_checkin(ctx: Ctx, event: Event) -> Step: if intake_enabled(): # the intake diary (#53, feature flag) replaces this check-in when released return await intake.checkin(ctx, event) return await start_checkin(ctx, from_menu=False) @trigger("button:menu:checkin", replace=True) async def _menu_checkin(ctx: Ctx, event: Event) -> Step: if intake_enabled(): return await intake.checkin(ctx, event) return await start_checkin(ctx, from_menu=True) @trigger("button:menu:visit", replace=True) async def _visit(ctx: Ctx, event: Event) -> Step: # «Записаться раньше» (rule action suggest_visit): no online booking yet — the request is a fact for lane B # (#38) and the dashboard, like `contact_requests`. No alert is created here (ADR-007). requests = [*(ctx.data.get("visit_requests") or []), {"ts": ctx.now.isoformat()}][-20:] return Step( replies=[Reply(text=load_daily_texts().visit.requested)], data={"visit_requests": requests}, notify_monitoring=True, ) @trigger("text:idle", replace=True) async def _idle_text(ctx: Ctx, event: Event) -> Step: """Free text outside the check-in: drug question → instruction excerpt; side effects / quit intent / alcohol → today's check-in (the rules react); another question → clinical guidelines (knowledge base); otherwise the neutral support phrase with the menu.""" assert ctx.text is not None facts = await ctx.gateway.extract_checkin(ctx.text) if facts is None: text = await knowledge_answer(ctx, explicit=False) if looks_like_question(ctx.text) else None return Step(replies=[menu_reply(ctx, text or await ctx.gateway.phrase("support"))]) fields: dict[str, Any] = {} if facts.quit_intent: fields["quit_intent"] = True if facts.alcohol is not None and facts.alcohol != "none": today = await checkin_of(ctx, ctx.now.date()) if today is None or today.alcohol_bucket in (None, "none"): fields["alcohol_bucket"] = facts.alcohol text: str | None = None if facts.side_effects: codes = [str(c) for c in facts.side_effects] await record_side_effects(ctx, codes) if not facts.quit_intent: text = (await side_effect_reply(ctx, codes)).text if fields: await save_checkin(ctx, **fields) if text is None and facts.med_question and not facts.quit_intent: text = await answer_question(ctx, await current_prescription(ctx)) if text is None and not fields and looks_like_question(ctx.text): text = await knowledge_answer(ctx, explicit=False) # general question → clinical guidelines (#64) if text is None: text = await ctx.gateway.phrase("support") return Step(replies=[menu_reply(ctx, text)], notify_monitoring=bool(fields or facts.side_effects)) # ------------------------------------------------------------------------------------------- scale intros def _scale_intro(ctx: Ctx, scale_id: str, scheduled: bool, return_state: str) -> str | None: """Friendly intro of a scheduled (or rule-triggered) scale. Onboarding sessions (#36) have their own texts.""" if ".onboarding" in return_state: return None intros = load_daily_texts().scale_intros return (intros.scheduled if scheduled else intros.unscheduled).get(scale_id) for _scale_id in load_daily_texts().settings.intro_scales: register_scale_intro(_scale_id, _scale_intro)