ElebobaandClaude Opus 5.5
31585c7f35
CI / ruff · pytest (pull_request) Successful in 2m41s
- Idle chat: all messages go to the agent (greetings, side-effect and substitute questions included); check-in signals are still recorded silently, quit intent keeps its rule reaction. Seen live: «какие будут побочки» was logged as a complaint («Спасибо, отметил»), «привет» got a canned phrase. - Agent prompt: warm short replies to small talk; substitutes → only the doctor changes the drug. - ЕМИАС demo persona button creates a private copy too (starting from consent + questionnaires, subject to the screening switch); the web app drops a stored session of the old shared demo account. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
485 lines
20 KiB
Python
485 lines
20 KiB
Python
"""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
|
||
from app.llm import agent
|
||
|
||
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:
|
||
smart = await agent.answer_chat(ctx.token, ctx.text)
|
||
if smart is not None:
|
||
return Step(replies=[menu_reply(ctx, agent.render(smart))])
|
||
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 not facts.quit_intent:
|
||
# Chat first: whatever the patient writes in idle gets the agent's answer; check-in signals above are
|
||
# recorded silently for the doctor. Quit intent keeps its rule reaction (explanation + doctor notified).
|
||
smart = await agent.answer_chat(ctx.token, ctx.text)
|
||
if smart is not None:
|
||
return Step(
|
||
replies=[menu_reply(ctx, agent.render(smart))], notify_monitoring=bool(fields or facts.side_effects)
|
||
)
|
||
if text is None and facts.med_question and not facts.quit_intent:
|
||
text = await answer_question(ctx, await current_prescription(ctx))
|
||
# A question is answered even if it also mentioned alcohol/side effects (those are recorded above).
|
||
# Quit intent gets the rule reaction (delayed-onset explanation + doctor notified) — no second answer.
|
||
if text is None and not facts.quit_intent 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)
|