Files
PTSRaiAgent/app/dialog/flows/intake.py
T
Arxip222andClaude Opus 5.5 44814b7293 fix(clinical): red flags, one crisis rule, callback after a crisis, one table of numbers
Clinical review of the questionnaires (ADR-013):
- Red-flag block first in stage 1 (config/instruments/red_flags.yaml); any «Да» stops the
  questionnaire with an acute alert and the help screen of its category.
- One crisis rule: the crisis screen opens at a HIGH C-SSRS risk only; moderate stops and goes
  to a psychiatrist (24 h as a guide), low is a flag with the safety plan. PHQ-9 item 9 with a
  negative screener is a discrepancy flag; without a completed screener it asks the C-SSRS at once.
  config/questionnaire.yaml: cssrs_positive_policy graded (C01).
- Every crisis leaves a callback task for the doctor on duty (post_crisis_call, one per patient)
  and the app asks how the person is the next day, the safety plan one tap away.
- Routing: risk and flags first, scales second; PCL-5 >= 50 no longer means a priority
  appointment (pcl5_severe is moderate on the dashboard too).
- Crisis numbers live only in config/texts/crisis.yaml; texts and the agent prompt refer to them
  via {phone:<key>}. Fixes the children's helpline offered to adults by the chat agent.

Refs #92

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-04 16:21:16 +03:00

511 lines
22 KiB
Python

"""Versioned, resumable questionnaires. Disabled until the clinical release gate opens."""
import math
from dataclasses import asdict, replace
from zoneinfo import ZoneInfo
from app.core import interpretation
from app.core.intake_questions import (
DIARY_KINDS,
Question,
content,
cssrs_answers,
cssrs_since_contact_question,
diary_questions,
group_answers,
next_intake,
red_flag_category,
scale_question,
texts,
)
from app.core.intake_runtime import diary_flags, intake_enabled, load_runtime
from app.core.questionnaire import alcohol_expansion, load_policy, score_alcohol, score_cssrs_recent
from app.core.scales import load_scale
from app.core.types import AlertLevel
from app.db import repo
from app.db.intake import QuestionnaireRecord, QuestionnaireTask
from app.dialog.contracts import Button, Event, Reply
from app.dialog.crisis import crisis_open, open_crisis
from app.dialog.fsm import Step, on_enter, state, trigger
from app.services import questionnaire as service
from app.services.followup import schedule_post_crisis_call
def allowed(ctx):
return ctx.data.get("consent") is True and ctx.data.get("adult") is True and intake_enabled()
def side_effects_question():
"""FIBSER-style tolerability question added to the full scales while a medication is active («Этап 2», p.3)."""
return Question("side_effects", texts()["side_effects"], ((1, "Да"), (0, "Нет")), "past_week")
def next_question(row):
a = row.answers
if row.kind == "intake":
return next_intake(a, load_policy(), row.details.get("sex"))
if row.kind == "cssrs_since_contact":
return cssrs_since_contact_question(a)
if row.kind in DIARY_KINDS:
part = row.kind.removeprefix("diary_")
for q in diary_questions(part, row.details["modules"], day_index=row.details.get("day_index", 0)):
if q.id in a:
continue
if q.id == "medication_reason" and a.get("medication") != "no":
continue
return q
return None
if row.kind in ("phq9", "pcl5", "rus_audit_s"):
sid = row.kind
if sid == "rus_audit_s":
short = score_alcohol(
{k: a[k] for k in load_scale(sid).item_ids if k in a}, short=True, sex=row.details.get("sex")
)
if alcohol_expansion(short, load_policy()):
sid = "rus_audit"
for item in load_scale(sid).item_ids:
if item == "rus_audit_2" and a.get("rus_audit_1") == 0:
continue
if item not in a:
q = scale_question(sid, item)
if sid == "pcl5" and row.details["weekly_pcl"]:
q = replace(q, text=q.text.replace("последний месяц", "последнюю неделю"), period="past_week")
return q
if row.details.get("medication_active") and "side_effects" not in a:
return side_effects_question()
return None
def question_reply(row, q):
t = texts()
# Nonce + item + option index rejects old/replayed callbacks without exposing raw answers.
prefix = f"iq:{row.id[:12]}:{q.id}:"
buttons = [[Button(label=label, data=prefix + str(i))] for i, (_, label) in enumerate(q.options)]
if q.optional:
buttons.append([Button(label=t["comment_skip"], data=prefix + "skip")])
buttons.extend([[Button(label=t["help"], data="intake:help")], [Button(label=t["pause"], data="intake:pause")]])
return Reply(text=q.text + "\n\n" + t["as_of"].format(day=row.day.isoformat()), buttons=buttons, kind="question")
async def urgent(ctx, row=None, *, code="help_now", categories=None):
if crisis_open(ctx.data):
return open_crisis(ctx, categories or [], source="questionnaire:" + code)
profile = await ctx.profile()
if profile:
profile.enhanced_monitoring = True
flags = list(service.decision(row).flags) if row else []
if row and (row.answers.get("dphq_9", 0) > 0 or row.answers.get("phq9_9", 0) > 0):
flags.append("cssrs_since_contact_required")
if row:
await service.finish(ctx, row, "interrupted", extra={"stop_reason": code, "flags": flags})
await service.alert(
ctx.session,
ctx.token,
f"{row.id if row else ctx.now.isoformat()}:{code}",
code,
ctx.now,
level=AlertLevel.ACUTE,
params={"record_id": row.id if row else None, "flags": flags},
)
await schedule_post_crisis_call(ctx.session, ctx.token, ctx.now, source="questionnaire:" + code)
step = open_crisis(ctx, categories or [], source="questionnaire:" + code)
step.data["questionnaire_record"] = None
step.data["intake_review_required"] = True
for reply in step.replies:
reply.buttons.append([Button(label=ctx.texts.safety_plan.title.rstrip(":"), data="safety_plan")])
return step
async def item9_discrepancy(ctx, row):
profile = await ctx.profile()
if profile:
profile.enhanced_monitoring = True
await service.alert(
ctx.session,
ctx.token,
row.id + ":phq9_cssrs_discrepancy",
"phq9_cssrs_discrepancy",
ctx.now,
level=AlertLevel.HIGH,
params={"record_id": row.id},
)
row.details = {**row.details, "item9_flagged": True}
async def item9_assessment(ctx, row):
"""Thoughts of death in PHQ-9 item 9 → the C-SSRS screener at once; the screener's level decides (graded)."""
await service.alert(
ctx.session,
ctx.token,
row.id + ":item9_assessment",
"phq9_item9_cssrs",
ctx.now,
level=AlertLevel.HIGH,
params={"record_id": row.id},
)
await service.finish(ctx, row, "complete", extra={"stop_reason": "phq9_item9_cssrs"})
profile = await ctx.profile()
if profile:
profile.enhanced_monitoring = True
task = await service.queue_task(ctx.session, ctx.token, "cssrs_since_contact", row.id + ":cssrs", ctx.now)
return await start(ctx, "cssrs_since_contact", task)
async def manual_review(ctx, row, code):
await service.finish(ctx, row, "review_required", extra={"stop_reason": code})
await service.alert(
ctx.session, ctx.token, f"{row.id}:{code}", code, ctx.now, level=AlertLevel.HIGH, params={"record_id": row.id}
)
return Step(
replies=[
Reply(
text=texts()["missing_followup"] if code == "cssrs_since_contact_required" else texts()["review"],
buttons=[[Button(label=texts()["help"], data="intake:help")]],
kind="info",
)
],
next_state="idle",
data={"questionnaire_record": None, "intake_review_required": True},
)
async def advance(ctx, row):
a = row.answers
c = score_cssrs_recent(cssrs_answers(a))
policy = load_policy()
# Red flags (asked first): any «Да» stops the questionnaire with the help screen of its category.
for item_id, value in a.items():
if item_id.startswith("rf_") and value == 1:
category = red_flag_category(item_id)
return await urgent(ctx, row, code="red_flag_" + item_id[3:], categories=[category] if category else [])
# One crisis rule for the whole app (ADR-013): the crisis screen opens at a HIGH C-SSRS risk only. Moderate →
# stop + psychiatrist (about 24 h); low → flag, safety plan, continue.
if c.risk_lower_bound == "high":
return await urgent(ctx, row, code="suicide_signal")
if a.get("dphq_9", 0) > 0 and row.kind in DIARY_KINDS:
# Diary depression module, item 9 ≥1 → the C-SSRS right away («Этап 2»); its risk level decides.
return await item9_assessment(ctx, row)
if a.get("phq9_9", 0) > 0 and not c.complete:
# Item 9 without a completed screener (only in a scheduled PHQ-9): assess the risk right away.
return await item9_assessment(ctx, row)
if a.get("phq9_9", 0) > 0 and not row.details.get("item9_flagged"):
# Item 9 ≥1 with a negative / low C-SSRS = discrepancy (routing table): flag to the doctor, safety plan.
await item9_discrepancy(ctx, row)
if c.positive and policy.cssrs_positive_policy == "stop_any":
return await urgent(ctx, row, code="cssrs_positive")
if c.conflicts:
return await manual_review(ctx, row, "cssrs_conflict")
# Finish the risk questions before any routine question; a moderate result stops here.
if c.complete and c.risk == "moderate":
return await manual_review(ctx, row, "urgent_psychiatrist")
if c.complete and c.risk == "low":
profile = await ctx.profile()
if profile:
profile.enhanced_monitoring = True
await service.alert(
ctx.session,
ctx.token,
row.id + ":cssrs_low",
"cssrs_low_safety_plan",
ctx.now,
params={"record_id": row.id},
)
try:
q = next_question(row)
# Validate preserved short answers against a newly answered full-form item 1.
score_alcohol(group_answers(a, "rus_audit"), short=False, sex=row.details.get("sex"))
except ValueError:
return await manual_review(ctx, row, "answer_conflict")
if q:
replies = [question_reply(row, q)]
if ((c.complete and c.risk == "low") or row.details.get("item9_flagged")) and not row.details.get("plan_shown"):
from app.dialog.actions import safety_plan_reply
replies.insert(0, await safety_plan_reply(ctx))
row.details = {**row.details, "plan_shown": True}
return Step(replies=replies, next_state="intake.collect", data={"questionnaire_record": row.id})
extra = {}
if a.get("side_effects") == 1:
await service.alert(
ctx.session,
ctx.token,
row.id + ":side_effects_review",
"side_effects_review",
ctx.now,
params={"record_id": row.id},
)
if row.kind == "intake":
result = service.decision(row)
if result.status != "ready":
return await manual_review(ctx, row, "intake_incomplete_review")
extra = {"decision": asdict(result), "totals": service.summary(row)}
if a.get("recent_trauma") == 1:
extra["recent_trauma"] = True
extra["interpretation"] = interpretation.intake_interpretation(extra["totals"], a, result, policy)
await apply_modules(ctx, extra["interpretation"]["modules"])
if result.primary_route != "self_help" or result.flags or result.parallel_routes or a.get("recent_trauma"):
await service.alert(
ctx.session,
ctx.token,
row.id + ":route",
"intake_route_review",
ctx.now,
params={"record_id": row.id, **extra},
)
elif row.kind == "cssrs_since_contact":
extra["totals"] = {"cssrs_recent": asdict(c)}
extra["interpretation"] = interpretation.cssrs_interpretation(c.risk)
elif row.kind in DIARY_KINDS:
# Morning and evening answers of one date form one diary day.
history: dict = {}
for r in await service.records(ctx.session, ctx.token):
if r.kind in DIARY_KINDS and r.status == "complete" and r.id != row.id:
history.setdefault(r.day, {}).update(r.answers)
history.setdefault(row.day, {}).update(a)
flags = diary_flags(history, row.day, medication_active=row.details.get("medication_active", False))
extra["flags"] = flags
for flag in flags:
await service.alert(
ctx.session, ctx.token, row.id + ":" + flag, flag, ctx.now, params={"record_id": row.id}
)
if flag in DIARY_SCALE_TASKS:
sid = DIARY_SCALE_TASKS[flag]
await service.queue_task(ctx.session, ctx.token, sid, row.id + ":" + sid, ctx.now)
extra["interpretation"] = interpretation.diary_interpretation(row.kind.removeprefix("diary_"), flags)
else:
extra["totals"] = service.summary(row)
runtime = load_runtime()
if row.kind in ("phq9", "pcl5"):
sid = row.kind
total = extra["totals"][sid]["total"]
# Only PHQ-9 has a priority threshold; PCL-5 alone never makes an appointment a priority.
priority = getattr(policy, sid + "_priority_at", None)
specialist = getattr(policy, sid + "_specialist_at")
extra["route"] = (
"priority_psychiatrist"
if priority is not None and total >= priority
else ("psychiatrist_or_psychotherapist" if total >= specialist else "review_with_treating_clinician")
)
earlier = [
r
for r in await service.records(ctx.session, ctx.token)
if r.id != row.id
and r.status == "complete"
and r.version == row.version
and r.details.get("totals", {}).get(sid, {}).get("complete")
and (sid != "pcl5" or r.details.get("weekly_pcl", False) == row.details.get("weekly_pcl", False))
]
prior = earlier[-1] if earlier else None
if prior:
change = total - prior.details["totals"][sid]["total"]
extra["delta"] = {
"baseline_record": prior.id,
"value": change,
"retriage": change >= getattr(runtime, sid + "_retriage_delta"),
}
extra["interpretation"] = interpretation.scale_interpretation(
sid,
extra["totals"],
baseline=earlier[0].details["totals"][sid]["total"] if earlier else None,
previous=prior.details["totals"][sid]["total"] if prior else None,
policy=policy,
retriage_delta=getattr(runtime, sid + "_retriage_delta"),
)
if row.kind == "rus_audit_s" and not row.details.get("sex"):
return await manual_review(ctx, row, "alcohol_sex_unknown")
if row.kind == "rus_audit_s":
extra["interpretation"] = interpretation.scale_interpretation(
"rus_audit_s", extra["totals"], baseline=None, previous=None, policy=policy, retriage_delta=0
)
if extra["totals"]["rus_audit"].get("zone") == 4:
extra["parallel_routes"] = ["narcologist"]
retriage = extra.get("delta", {}).get("retriage", False)
await service.alert(
ctx.session,
ctx.token,
row.id + ":followup",
"followup_retriage_required" if retriage else "followup_review",
ctx.now,
level=AlertLevel.HIGH if retriage or extra.get("route") == "priority_psychiatrist" else AlertLevel.MODERATE,
params={"record_id": row.id, **extra},
)
await service.finish(ctx, row, "complete", extra=extra)
shown = interpretation.render(extra["interpretation"]) if extra.get("interpretation") else texts()["done"]
if row.kind == "cssrs_since_contact" and not await done_today(ctx, "diary_evening"):
# The risk questions open the evening diary; continue straight into it.
step = await start(ctx, "diary_evening")
step.replies.insert(0, Reply(text=shown, kind="info"))
return step
from app.dialog.flows.core import menu_reply
replies = [menu_reply(ctx, shown)]
if ((c.complete and c.risk == "low") or row.details.get("item9_flagged")) and not row.details.get("plan_shown"):
# The safety plan is never skipped because the risk signal came with the last question.
from app.dialog.actions import safety_plan_reply
replies.insert(0, await safety_plan_reply(ctx))
return Step(replies=replies, next_state="idle", data={"questionnaire_record": None})
# Diary flags that bring a full scale forward («Этап 2», p.4).
DIARY_SCALE_TASKS = {
"phq9_due_mood": "phq9",
"phq9_due_depression": "phq9",
"pcl5_due_ptsd": "pcl5",
"rus_audit_due_alcohol": "rus_audit_s",
}
async def apply_modules(ctx, modules):
"""Stage 1 result → diary modules of stage 2. Without a doctor the screening defines them; a doctor's modules
(diagnosis from ЕМИАС) are kept and only extended."""
profile = await ctx.profile()
if profile is None:
await repo.upsert_profile(ctx.session, token=ctx.token, doctor_id="", modules=list(modules))
return
current = list(profile.modules or [])
merged = [*current, *(m for m in modules if m not in current)] if profile.doctor_id else list(modules)
await repo.upsert_profile(ctx.session, token=ctx.token, modules=merged)
def local_day(ctx):
return ctx.now.astimezone(ZoneInfo(load_runtime().timezone)).date()
async def done_today(ctx, kind):
day = local_day(ctx)
return any(
r.kind == kind and r.day == day and r.status == "complete"
for r in await service.records(ctx.session, ctx.token)
)
async def start(ctx, kind="intake", task=None):
if not allowed(ctx) or crisis_open(ctx.data):
if task:
return Step()
return Step(replies=[Reply(text=texts()["unavailable"], kind="info")])
existing = await service.records(ctx.session, ctx.token)
active = next((r for r in reversed(existing) if r.status in ("active", "paused")), None)
if active:
if task and task.record_id != active.id:
return Step()
if (
active.policy_version != load_policy().version
or active.version != content()["version"]
or active.details.get("runtime_version") != load_runtime().version
):
return await manual_review(ctx, active, "questionnaire_version_changed")
active.status = "active"
return await advance(ctx, active)
if ctx.data.get("intake_review_required") and kind != "cssrs_since_contact":
if task:
return Step()
return Step(replies=[Reply(text=texts()["review"], kind="info")])
# A manual start takes over the scheduled task of the same form, so the task does not stay outstanding.
task = task or await service.outstanding_task(ctx.session, ctx.token, kind)
row = await service.create_record(ctx, kind, task=task)
return await advance(ctx, row)
@on_enter("intake.start")
async def enter(ctx):
return await start(ctx)
@trigger("button:intake:start")
@trigger("button:intake:resume")
async def start_button(ctx, event):
return await start(ctx)
@trigger("button:menu:checkin", replace=True)
@trigger("system:daily_checkin", replace=True)
async def checkin(ctx, event):
if not intake_enabled():
return Step(replies=[Reply(text=ctx.texts.idle.stub_checkin)])
if event.params.get("questionnaire_task"):
task = await ctx.session.get(QuestionnaireTask, event.params["questionnaire_task"])
if task is None or task.token != ctx.token or task.status not in ("pending", "active"):
return Step()
return await start(ctx, task.kind, task)
existing = await service.records(ctx.session, ctx.token)
if not any(r.kind == "intake" and r.status == "complete" for r in existing):
return await start(ctx)
return await next_diary(ctx)
async def next_diary(ctx):
"""The part of today's diary that is still open: morning first, then evening. The evening opens with the C-SSRS
«since last contact» when the profile is under suicide-risk monitoring («Этап 2», module of suicide risk)."""
day = local_day(ctx)
done = {r.kind for r in await service.records(ctx.session, ctx.token) if r.day == day and r.status == "complete"}
for kind in DIARY_KINDS:
if kind in done:
continue
profile = await ctx.profile()
if kind == "diary_evening" and profile and profile.enhanced_monitoring and "cssrs_since_contact" not in done:
risk_task = await service.queue_task(ctx.session, ctx.token, "cssrs_since_contact", f"cssrs:{day}", ctx.now)
return await start(ctx, "cssrs_since_contact", risk_task)
return await start(ctx, kind)
return Step(replies=[Reply(text=texts()["diary_done"], kind="info")])
@trigger("button:diary:next")
async def diary_button(ctx, event):
return await next_diary(ctx)
@state("intake")
async def collect(ctx, event: Event):
if not allowed(ctx):
return Step(replies=[Reply(text=texts()["unavailable"], kind="info")], next_state="idle")
row = await ctx.session.get(QuestionnaireRecord, ctx.data.get("questionnaire_record", ""))
if row is None or row.token != ctx.token or row.status != "active":
return Step(next_state="idle")
if (
row.policy_version != load_policy().version
or row.version != content()["version"]
or row.details.get("runtime_version") != load_runtime().version
):
return await manual_review(ctx, row, "questionnaire_version_changed")
if event.type == "system":
task_id = event.params.get("questionnaire_task")
if task_id and row.details.get("task_id") == task_id:
return await advance(ctx, row)
return Step()
if event.button == "intake:pause":
row.status = "paused"
return Step(
replies=[Reply(text=texts()["paused"], buttons=[[Button(label=texts()["resume"], data="intake:resume")]])],
next_state="idle",
)
q = next_question(row)
if q is None:
return await advance(ctx, row)
prefix = f"iq:{row.id[:12]}:{q.id}:"
value = None
valid = False
if event.type == "button" and (event.button or "").startswith(prefix):
index = event.button[len(prefix) :]
if index == "skip" and q.optional:
valid = True
elif index.isdigit() and int(index) < len(q.options):
value, valid = q.options[int(index)][0], True
elif event.type == "text" and q.numeric:
try:
value = float(str(ctx.text).replace(",", "."))
valid = math.isfinite(value) and 0 <= value <= q.maximum
except ValueError:
pass
elif event.type == "text" and q.optional:
value, valid = str(ctx.text)[:2000], True
if not valid:
hint = "invalid" if not q.numeric else ("numeric_portions" if q.id == "alcohol_portions" else "numeric")
return Step(replies=[Reply(text=texts()[hint]), question_reply(row, q)])
service.save_answer(row, q, value, ctx.now)
if row.kind == "cssrs_since_contact" and q.id == "cssrs_q6_lifetime":
# Behaviour since the last contact is recent by definition (see cssrs_since_contact_question).
service.save_answer(row, replace(q, id="cssrs_q6_recent"), value, ctx.now, derived=True)
return await advance(ctx, row)